diff options
Diffstat (limited to 'test/ext/declarative/test_inheritance.py')
| -rw-r--r-- | test/ext/declarative/test_inheritance.py | 1585 |
1 files changed, 892 insertions, 693 deletions
diff --git a/test/ext/declarative/test_inheritance.py b/test/ext/declarative/test_inheritance.py index a453bc165..6cf0073b2 100644 --- a/test/ext/declarative/test_inheritance.py +++ b/test/ext/declarative/test_inheritance.py @@ -1,16 +1,34 @@ - -from sqlalchemy.testing import eq_, le_, assert_raises, \ - assert_raises_message, is_, is_true, is_false +from sqlalchemy.testing import ( + eq_, + le_, + assert_raises, + assert_raises_message, + is_, + is_true, + is_false, +) from sqlalchemy.ext import declarative as decl import sqlalchemy as sa from sqlalchemy import testing from sqlalchemy import Integer, String, ForeignKey from sqlalchemy.testing.schema import Table, Column -from sqlalchemy.orm import relationship, create_session, class_mapper, \ - configure_mappers, clear_mappers, \ - polymorphic_union, deferred, Session, mapper -from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \ - ConcreteBase, has_inherited_table +from sqlalchemy.orm import ( + relationship, + create_session, + class_mapper, + configure_mappers, + clear_mappers, + polymorphic_union, + deferred, + Session, + mapper, +) +from sqlalchemy.ext.declarative import ( + declared_attr, + AbstractConcreteBase, + ConcreteBase, + has_inherited_table, +) from sqlalchemy.testing import fixtures, mock from test.orm.test_events import _RemoveListeners @@ -18,7 +36,6 @@ Base = None class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults): - def setup(self): global Base Base = decl.declarative_base(testing.db) @@ -30,119 +47,133 @@ class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults): class DeclarativeInheritanceTest(DeclarativeTestBase): - def test_we_must_copy_mapper_args(self): - class Person(Base): - __tablename__ = 'people' + __tablename__ = "people" id = Column(Integer, primary_key=True) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator, - 'polymorphic_identity': 'person'} + discriminator = Column("type", String(50)) + __mapper_args__ = { + "polymorphic_on": discriminator, + "polymorphic_identity": "person", + } class Engineer(Person): primary_language = Column(String(50)) - assert 'inherits' not in Person.__mapper_args__ + assert "inherits" not in Person.__mapper_args__ assert class_mapper(Engineer).polymorphic_identity is None assert class_mapper(Engineer).polymorphic_on is Person.__table__.c.type def test_we_must_only_copy_column_mapper_args(self): - class Person(Base): - __tablename__ = 'people' + __tablename__ = "people" id = Column(Integer, primary_key=True) a = Column(Integer) b = Column(Integer) c = Column(Integer) d = Column(Integer) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator, - 'polymorphic_identity': 'person', - 'version_id_col': 'a', - 'column_prefix': 'bar', - 'include_properties': ['id', 'a', 'b'], - } - assert class_mapper(Person).version_id_col == 'a' - assert class_mapper(Person).include_properties == set(['id', 'a', 'b']) + discriminator = Column("type", String(50)) + __mapper_args__ = { + "polymorphic_on": discriminator, + "polymorphic_identity": "person", + "version_id_col": "a", + "column_prefix": "bar", + "include_properties": ["id", "a", "b"], + } - def test_custom_join_condition(self): + assert class_mapper(Person).version_id_col == "a" + assert class_mapper(Person).include_properties == set(["id", "a", "b"]) + def test_custom_join_condition(self): class Foo(Base): - __tablename__ = 'foo' - id = Column('id', Integer, primary_key=True) + __tablename__ = "foo" + id = Column("id", Integer, primary_key=True) class Bar(Foo): - __tablename__ = 'bar' - bar_id = Column('id', Integer, primary_key=True) - foo_id = Column('foo_id', Integer) - __mapper_args__ = {'inherit_condition': foo_id == Foo.id} + __tablename__ = "bar" + bar_id = Column("id", Integer, primary_key=True) + foo_id = Column("foo_id", Integer) + __mapper_args__ = {"inherit_condition": foo_id == Foo.id} # compile succeeds because inherit_condition is honored configure_mappers() def test_joined(self): - class Company(Base, fixtures.ComparableEntity): - __tablename__ = 'companies' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - employees = relationship('Person') + __tablename__ = "companies" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + employees = relationship("Person") class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - company_id = Column('company_id', Integer, - ForeignKey('companies.id')) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + company_id = Column( + "company_id", Integer, ForeignKey("companies.id") + ) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) - primary_language = Column('primary_language', String(50)) + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) + primary_language = Column("primary_language", String(50)) class Manager(Person): - __tablename__ = 'managers' - __mapper_args__ = {'polymorphic_identity': 'manager'} - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) - golf_swing = Column('golf_swing', String(50)) + __tablename__ = "managers" + __mapper_args__ = {"polymorphic_identity": "manager"} + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) + golf_swing = Column("golf_swing", String(50)) Base.metadata.create_all() sess = create_session() c1 = Company( - name='MegaCorp, Inc.', + name="MegaCorp, Inc.", employees=[ - Engineer(name='dilbert', primary_language='java'), - Engineer(name='wally', primary_language='c++'), - Manager(name='dogbert', golf_swing='fore!')]) + Engineer(name="dilbert", primary_language="java"), + Engineer(name="wally", primary_language="c++"), + Manager(name="dogbert", golf_swing="fore!"), + ], + ) - c2 = Company(name='Elbonia, Inc.', - employees=[Engineer(name='vlad', - primary_language='cobol')]) + c2 = Company( + name="Elbonia, Inc.", + employees=[Engineer(name="vlad", primary_language="cobol")], + ) sess.add(c1) sess.add(c2) sess.flush() sess.expunge_all() - eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). - any(Engineer.primary_language - == 'cobol')).first(), c2) + eq_( + sess.query(Company) + .filter( + Company.employees.of_type(Engineer).any( + Engineer.primary_language == "cobol" + ) + ) + .first(), + c2, + ) # ensure that the Manager mapper was compiled with the Manager id # column as higher priority. this ensures that "Manager.id" @@ -151,7 +182,7 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): eq_( Manager.id.property.columns, - [Manager.__table__.c.id, Person.__table__.c.id] + [Manager.__table__.c.id, Person.__table__.c.id], ) # assert that the "id" column is available without a second @@ -161,120 +192,130 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): sess.expunge_all() def go(): - assert sess.query(Manager).filter(Manager.name == 'dogbert' - ).one().id + assert ( + sess.query(Manager).filter(Manager.name == "dogbert").one().id + ) + self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): - assert sess.query(Person).filter(Manager.name == 'dogbert' - ).one().id + assert ( + sess.query(Person).filter(Manager.name == "dogbert").one().id + ) self.assert_sql_count(testing.db, go, 1) def test_add_subcol_after_the_fact(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) - Engineer.primary_language = Column('primary_language', - String(50)) + Engineer.primary_language = Column("primary_language", String(50)) Base.metadata.create_all() sess = create_session() - e1 = Engineer(primary_language='java', name='dilbert') + e1 = Engineer(primary_language="java", name="dilbert") sess.add(e1) sess.flush() sess.expunge_all() - eq_(sess.query(Person).first(), - Engineer(primary_language='java', name='dilbert')) + eq_( + sess.query(Person).first(), + Engineer(primary_language="java", name="dilbert"), + ) def test_add_parentcol_after_the_fact(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) - Person.name = Column('name', String(50)) + Person.name = Column("name", String(50)) Base.metadata.create_all() sess = create_session() - e1 = Engineer(primary_language='java', name='dilbert') + e1 = Engineer(primary_language="java", name="dilbert") sess.add(e1) sess.flush() sess.expunge_all() - eq_(sess.query(Person).first(), - Engineer(primary_language='java', name='dilbert')) + eq_( + sess.query(Person).first(), + Engineer(primary_language="java", name="dilbert"), + ) def test_add_sub_parentcol_after_the_fact(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) class Admin(Engineer): - __tablename__ = 'admins' - __mapper_args__ = {'polymorphic_identity': 'admin'} + __tablename__ = "admins" + __mapper_args__ = {"polymorphic_identity": "admin"} workstation = Column(String(50)) - id = Column('id', Integer, ForeignKey('engineers.id'), - primary_key=True) + id = Column( + "id", Integer, ForeignKey("engineers.id"), primary_key=True + ) - Person.name = Column('name', String(50)) + Person.name = Column("name", String(50)) Base.metadata.create_all() sess = create_session() - e1 = Admin(primary_language='java', name='dilbert', - workstation='foo') + e1 = Admin(primary_language="java", name="dilbert", workstation="foo") sess.add(e1) sess.flush() sess.expunge_all() - eq_(sess.query(Person).first(), - Admin(primary_language='java', name='dilbert', workstation='foo')) + eq_( + sess.query(Person).first(), + Admin(primary_language="java", name="dilbert", workstation="foo"), + ) def test_subclass_mixin(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class MyMixin(object): @@ -282,11 +323,12 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): class Engineer(MyMixin, Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} - id = Column('id', Integer, ForeignKey('people.id'), - primary_key=True) - primary_language = Column('primary_language', String(50)) + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} + id = Column( + "id", Integer, ForeignKey("people.id"), primary_key=True + ) + primary_language = Column("primary_language", String(50)) assert class_mapper(Engineer).inherits is class_mapper(Person) @@ -294,69 +336,88 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): class Person(object): pass - person_table = Table('people', Base.metadata, - Column('id', Integer, primary_key=True), - Column('kind', String(50))) + person_table = Table( + "people", + Base.metadata, + Column("id", Integer, primary_key=True), + Column("kind", String(50)), + ) - mapper(Person, person_table, - polymorphic_on='kind', polymorphic_identity='person') + mapper( + Person, + person_table, + polymorphic_on="kind", + polymorphic_identity="person", + ) class SpecialPerson(Person): __abstract__ = True class Manager(SpecialPerson, Base): - __tablename__ = 'managers' + __tablename__ = "managers" id = Column(Integer, ForeignKey(Person.id), primary_key=True) - __mapper_args__ = { - 'polymorphic_identity': 'manager' - } + __mapper_args__ = {"polymorphic_identity": "manager"} from sqlalchemy import inspect + assert inspect(Manager).inherits is inspect(Person) - eq_(set(class_mapper(Person).class_manager), {'id', 'kind'}) - eq_(set(class_mapper(Manager).class_manager), {'id', 'kind'}) + eq_(set(class_mapper(Person).class_manager), {"id", "kind"}) + eq_(set(class_mapper(Manager).class_manager), {"id", "kind"}) def test_intermediate_unmapped_class_on_classical(self): class Person(object): pass - person_table = Table('people', Base.metadata, - Column('id', Integer, primary_key=True), - Column('kind', String(50))) + person_table = Table( + "people", + Base.metadata, + Column("id", Integer, primary_key=True), + Column("kind", String(50)), + ) - mapper(Person, person_table, - polymorphic_on='kind', polymorphic_identity='person') + mapper( + Person, + person_table, + polymorphic_on="kind", + polymorphic_identity="person", + ) class SpecialPerson(Person): pass class Manager(SpecialPerson, Base): - __tablename__ = 'managers' + __tablename__ = "managers" id = Column(Integer, ForeignKey(Person.id), primary_key=True) - __mapper_args__ = { - 'polymorphic_identity': 'manager' - } + __mapper_args__ = {"polymorphic_identity": "manager"} from sqlalchemy import inspect + assert inspect(Manager).inherits is inspect(Person) - eq_(set(class_mapper(Person).class_manager), {'id', 'kind'}) - eq_(set(class_mapper(Manager).class_manager), {'id', 'kind'}) + eq_(set(class_mapper(Person).class_manager), {"id", "kind"}) + eq_(set(class_mapper(Manager).class_manager), {"id", "kind"}) def test_class_w_invalid_multiple_bases(self): class Person(object): pass - person_table = Table('people', Base.metadata, - Column('id', Integer, primary_key=True), - Column('kind', String(50))) + person_table = Table( + "people", + Base.metadata, + Column("id", Integer, primary_key=True), + Column("kind", String(50)), + ) - mapper(Person, person_table, - polymorphic_on='kind', polymorphic_identity='person') + mapper( + Person, + person_table, + polymorphic_on="kind", + polymorphic_identity="person", + ) class DeclPerson(Base): - __tablename__ = 'decl_people' + __tablename__ = "decl_people" id = Column(Integer, primary_key=True) kind = Column(String(50)) @@ -365,48 +426,47 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): def go(): class Manager(SpecialPerson, DeclPerson): - __tablename__ = 'managers' - id = Column(Integer, - ForeignKey(DeclPerson.id), primary_key=True) - __mapper_args__ = { - 'polymorphic_identity': 'manager' - } + __tablename__ = "managers" + id = Column( + Integer, ForeignKey(DeclPerson.id), primary_key=True + ) + __mapper_args__ = {"polymorphic_identity": "manager"} assert_raises_message( sa.exc.InvalidRequestError, r"Class .*Manager.* has multiple mapped " r"bases: \[.*Person.*DeclPerson.*\]", - go + go, ) def test_with_undefined_foreignkey(self): - class Parent(Base): - __tablename__ = 'parent' - id = Column('id', Integer, primary_key=True) - tp = Column('type', String(50)) + __tablename__ = "parent" + id = Column("id", Integer, primary_key=True) + tp = Column("type", String(50)) __mapper_args__ = dict(polymorphic_on=tp) class Child1(Parent): - __tablename__ = 'child1' - id = Column('id', Integer, ForeignKey('parent.id'), - primary_key=True) - related_child2 = Column('c2', Integer, - ForeignKey('child2.id')) - __mapper_args__ = dict(polymorphic_identity='child1') + __tablename__ = "child1" + id = Column( + "id", Integer, ForeignKey("parent.id"), primary_key=True + ) + related_child2 = Column("c2", Integer, ForeignKey("child2.id")) + __mapper_args__ = dict(polymorphic_identity="child1") # no exception is raised by the ForeignKey to "child2" even # though child2 doesn't exist yet class Child2(Parent): - __tablename__ = 'child2' - id = Column('id', Integer, ForeignKey('parent.id'), - primary_key=True) - related_child1 = Column('c1', Integer) - __mapper_args__ = dict(polymorphic_identity='child2') + __tablename__ = "child2" + id = Column( + "id", Integer, ForeignKey("parent.id"), primary_key=True + ) + related_child1 = Column("c1", Integer) + __mapper_args__ = dict(polymorphic_identity="child2") sa.orm.configure_mappers() # no exceptions here @@ -419,31 +479,29 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): """ class Booking(Base): - __tablename__ = 'booking' + __tablename__ = "booking" id = Column(Integer, primary_key=True) class PlanBooking(Booking): - __tablename__ = 'plan_booking' - id = Column(Integer, ForeignKey(Booking.id), - primary_key=True) + __tablename__ = "plan_booking" + id = Column(Integer, ForeignKey(Booking.id), primary_key=True) # referencing PlanBooking.id gives us the column # on plan_booking, not booking class FeatureBooking(Booking): - __tablename__ = 'feature_booking' - id = Column(Integer, ForeignKey(Booking.id), - primary_key=True) - plan_booking_id = Column(Integer, - ForeignKey(PlanBooking.id)) + __tablename__ = "feature_booking" + id = Column(Integer, ForeignKey(Booking.id), primary_key=True) + plan_booking_id = Column(Integer, ForeignKey(PlanBooking.id)) - plan_booking = relationship(PlanBooking, - backref='feature_bookings') + plan_booking = relationship( + PlanBooking, backref="feature_bookings" + ) - assert FeatureBooking.__table__.c.plan_booking_id.\ - references(PlanBooking.__table__.c.id) + assert FeatureBooking.__table__.c.plan_booking_id.references( + PlanBooking.__table__.c.id + ) - assert FeatureBooking.__table__.c.id.\ - references(Booking.__table__.c.id) + assert FeatureBooking.__table__.c.id.references(Booking.__table__.c.id) def test_single_colsonbase(self): """test single inheritance where all the columns are on the base @@ -451,55 +509,71 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): class Company(Base, fixtures.ComparableEntity): - __tablename__ = 'companies' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - employees = relationship('Person') + __tablename__ = "companies" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + employees = relationship("Person") class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - company_id = Column('company_id', Integer, - ForeignKey('companies.id')) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - primary_language = Column('primary_language', String(50)) - golf_swing = Column('golf_swing', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + company_id = Column( + "company_id", Integer, ForeignKey("companies.id") + ) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + primary_language = Column("primary_language", String(50)) + golf_swing = Column("golf_swing", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __mapper_args__ = {"polymorphic_identity": "engineer"} class Manager(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} Base.metadata.create_all() sess = create_session() c1 = Company( - name='MegaCorp, Inc.', + name="MegaCorp, Inc.", employees=[ - Engineer(name='dilbert', primary_language='java'), - Engineer(name='wally', primary_language='c++'), - Manager(name='dogbert', golf_swing='fore!')]) + Engineer(name="dilbert", primary_language="java"), + Engineer(name="wally", primary_language="c++"), + Manager(name="dogbert", golf_swing="fore!"), + ], + ) - c2 = Company(name='Elbonia, Inc.', - employees=[Engineer(name='vlad', - primary_language='cobol')]) + c2 = Company( + name="Elbonia, Inc.", + employees=[Engineer(name="vlad", primary_language="cobol")], + ) sess.add(c1) sess.add(c2) sess.flush() sess.expunge_all() - eq_(sess.query(Person).filter(Engineer.primary_language - == 'cobol').first(), - Engineer(name='vlad')) - eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). - any(Engineer.primary_language - == 'cobol')).first(), c2) + eq_( + sess.query(Person) + .filter(Engineer.primary_language == "cobol") + .first(), + Engineer(name="vlad"), + ) + eq_( + sess.query(Company) + .filter( + Company.employees.of_type(Engineer).any( + Engineer.primary_language == "cobol" + ) + ) + .first(), + c2, + ) def test_single_colsonsub(self): """test single inheritance where the columns are local to their @@ -511,30 +585,32 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): class Company(Base, fixtures.ComparableEntity): - __tablename__ = 'companies' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - employees = relationship('Person') + __tablename__ = "companies" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + employees = relationship("Person") class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - company_id = Column(Integer, ForeignKey('companies.id')) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + company_id = Column(Integer, ForeignKey("companies.id")) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) class Manager(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} golf_swing = Column(String(50)) # we have here a situation that is somewhat unique. the Person @@ -549,31 +625,42 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): assert Person.__table__.c.primary_language is not None assert Engineer.primary_language is not None assert Manager.golf_swing is not None - assert not hasattr(Person, 'primary_language') - assert not hasattr(Person, 'golf_swing') - assert not hasattr(Engineer, 'golf_swing') - assert not hasattr(Manager, 'primary_language') + assert not hasattr(Person, "primary_language") + assert not hasattr(Person, "golf_swing") + assert not hasattr(Engineer, "golf_swing") + assert not hasattr(Manager, "primary_language") Base.metadata.create_all() sess = create_session() - e1 = Engineer(name='dilbert', primary_language='java') - e2 = Engineer(name='wally', primary_language='c++') - m1 = Manager(name='dogbert', golf_swing='fore!') - c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1]) - e3 = Engineer(name='vlad', primary_language='cobol') - c2 = Company(name='Elbonia, Inc.', employees=[e3]) + e1 = Engineer(name="dilbert", primary_language="java") + e2 = Engineer(name="wally", primary_language="c++") + m1 = Manager(name="dogbert", golf_swing="fore!") + c1 = Company(name="MegaCorp, Inc.", employees=[e1, e2, m1]) + e3 = Engineer(name="vlad", primary_language="cobol") + c2 = Company(name="Elbonia, Inc.", employees=[e3]) sess.add(c1) sess.add(c2) sess.flush() sess.expunge_all() - eq_(sess.query(Person).filter(Engineer.primary_language - == 'cobol').first(), - Engineer(name='vlad')) - eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). - any(Engineer.primary_language - == 'cobol')).first(), c2) - eq_(sess.query(Engineer).filter_by(primary_language='cobol' - ).one(), - Engineer(name='vlad', primary_language='cobol')) + eq_( + sess.query(Person) + .filter(Engineer.primary_language == "cobol") + .first(), + Engineer(name="vlad"), + ) + eq_( + sess.query(Company) + .filter( + Company.employees.of_type(Engineer).any( + Engineer.primary_language == "cobol" + ) + ) + .first(), + c2, + ) + eq_( + sess.query(Engineer).filter_by(primary_language="cobol").one(), + Engineer(name="vlad", primary_language="cobol"), + ) def test_single_cols_on_sub_base_of_joined(self): """test [ticket:3895]""" @@ -584,16 +671,12 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): id = Column(Integer, primary_key=True) type = Column(String) - __mapper_args__ = { - "polymorphic_on": type, - } + __mapper_args__ = {"polymorphic_on": type} class Contractor(Person): contractor_field = Column(String) - __mapper_args__ = { - "polymorphic_identity": "contractor", - } + __mapper_args__ = {"polymorphic_identity": "contractor"} class Employee(Person): __tablename__ = "employee" @@ -601,104 +684,115 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): id = Column(Integer, ForeignKey(Person.id), primary_key=True) class Engineer(Employee): - __mapper_args__ = { - "polymorphic_identity": "engineer", - } + __mapper_args__ = {"polymorphic_identity": "engineer"} configure_mappers() - is_false(hasattr(Person, 'contractor_field')) - is_true(hasattr(Contractor, 'contractor_field')) - is_false(hasattr(Employee, 'contractor_field')) - is_false(hasattr(Engineer, 'contractor_field')) + is_false(hasattr(Person, "contractor_field")) + is_true(hasattr(Contractor, "contractor_field")) + is_false(hasattr(Employee, "contractor_field")) + is_false(hasattr(Engineer, "contractor_field")) def test_single_cols_on_sub_to_joined(self): """test [ticket:3797]""" class BaseUser(Base): - __tablename__ = 'root' + __tablename__ = "root" id = Column(Integer, primary_key=True) row_type = Column(String) __mapper_args__ = { - 'polymorphic_on': row_type, - 'polymorphic_identity': 'baseuser' + "polymorphic_on": row_type, + "polymorphic_identity": "baseuser", } class User(BaseUser): - __tablename__ = 'user' + __tablename__ = "user" - __mapper_args__ = { - 'polymorphic_identity': 'user' - } + __mapper_args__ = {"polymorphic_identity": "user"} baseuser_id = Column( - Integer, ForeignKey('root.id'), primary_key=True) + Integer, ForeignKey("root.id"), primary_key=True + ) class Bat(Base): - __tablename__ = 'bat' + __tablename__ = "bat" id = Column(Integer, primary_key=True) class Thing(Base): - __tablename__ = 'thing' + __tablename__ = "thing" id = Column(Integer, primary_key=True) - owner_id = Column(Integer, ForeignKey('user.baseuser_id')) - owner = relationship('User') + owner_id = Column(Integer, ForeignKey("user.baseuser_id")) + owner = relationship("User") class SubUser(User): - __mapper_args__ = { - 'polymorphic_identity': 'subuser' - } + __mapper_args__ = {"polymorphic_identity": "subuser"} - sub_user_custom_thing = Column(Integer, ForeignKey('bat.id')) + sub_user_custom_thing = Column(Integer, ForeignKey("bat.id")) eq_( User.__table__.foreign_keys, User.baseuser_id.foreign_keys.union( - SubUser.sub_user_custom_thing.foreign_keys)) - is_true(Thing.owner.property.primaryjoin.compare( - Thing.owner_id == User.baseuser_id)) + SubUser.sub_user_custom_thing.foreign_keys + ), + ) + is_true( + Thing.owner.property.primaryjoin.compare( + Thing.owner_id == User.baseuser_id + ) + ) def test_single_constraint_on_sub(self): """test the somewhat unusual case of [ticket:3341]""" class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) __hack_args_one__ = sa.UniqueConstraint( - Person.name, primary_language) + Person.name, primary_language + ) __hack_args_two__ = sa.CheckConstraint( - Person.name != primary_language) - - uq = [c for c in Person.__table__.constraints - if isinstance(c, sa.UniqueConstraint)][0] - ck = [c for c in Person.__table__.constraints - if isinstance(c, sa.CheckConstraint)][0] + Person.name != primary_language + ) + + uq = [ + c + for c in Person.__table__.constraints + if isinstance(c, sa.UniqueConstraint) + ][0] + ck = [ + c + for c in Person.__table__.constraints + if isinstance(c, sa.CheckConstraint) + ][0] eq_( list(uq.columns), - [Person.__table__.c.name, Person.__table__.c.primary_language] + [Person.__table__.c.name, Person.__table__.c.primary_language], ) eq_( list(ck.columns), - [Person.__table__.c.name, Person.__table__.c.primary_language] + [Person.__table__.c.name, Person.__table__.c.primary_language], ) - @testing.skip_if(lambda: testing.against('oracle'), - "Test has an empty insert in it at the moment") + @testing.skip_if( + lambda: testing.against("oracle"), + "Test has an empty insert in it at the moment", + ) def test_columns_single_inheritance_conflict_resolution(self): """Test that a declared_attr can return the existing column and it will be ignored. this allows conditional columns to be added. @@ -706,8 +800,9 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): See [ticket:2472]. """ + class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column(Integer, primary_key=True) class Engineer(Person): @@ -717,8 +812,8 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): @declared_attr def target_id(cls): return cls.__table__.c.get( - 'target_id', - Column(Integer, ForeignKey('other.id'))) + "target_id", Column(Integer, ForeignKey("other.id")) + ) @declared_attr def target(cls): @@ -731,34 +826,31 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): @declared_attr def target_id(cls): return cls.__table__.c.get( - 'target_id', - Column(Integer, ForeignKey('other.id'))) + "target_id", Column(Integer, ForeignKey("other.id")) + ) @declared_attr def target(cls): return relationship("Other") class Other(Base): - __tablename__ = 'other' + __tablename__ = "other" id = Column(Integer, primary_key=True) is_( Engineer.target_id.property.columns[0], - Person.__table__.c.target_id + Person.__table__.c.target_id, ) is_( - Manager.target_id.property.columns[0], - Person.__table__.c.target_id + Manager.target_id.property.columns[0], Person.__table__.c.target_id ) # do a brief round trip on this Base.metadata.create_all() session = Session() o1, o2 = Other(), Other() - session.add_all([ - Engineer(target=o1), - Manager(target=o2), - Manager(target=o1) - ]) + session.add_all( + [Engineer(target=o1), Manager(target=o2), Manager(target=o1)] + ) session.commit() eq_(session.query(Engineer).first().target, o1) @@ -767,8 +859,9 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): #4352. """ + class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column(Integer, primary_key=True) target_id = Column(Integer, primary_key=True) @@ -780,8 +873,8 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): @declared_attr def target_id(cls): return cls.__table__.c.get( - 'target_id', - Column(Integer, primary_key=True)) + "target_id", Column(Integer, primary_key=True) + ) class Manager(Person): @@ -790,22 +883,22 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): @declared_attr def target_id(cls): return cls.__table__.c.get( - 'target_id', - Column(Integer, primary_key=True)) + "target_id", Column(Integer, primary_key=True) + ) is_( Engineer.target_id.property.columns[0], - Person.__table__.c.target_id + Person.__table__.c.target_id, ) is_( - Manager.target_id.property.columns[0], - Person.__table__.c.target_id + Manager.target_id.property.columns[0], Person.__table__.c.target_id ) def test_columns_single_inheritance_cascading_resolution_pk(self): """An additional test for #4352 in terms of the requested use case. """ + class TestBase(Base): __abstract__ = True @@ -813,14 +906,15 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): def id(cls): col_val = None if TestBase not in cls.__bases__: - col_val = cls.__table__.c.get('id') + col_val = cls.__table__.c.get("id") if col_val is None: col_val = Column(Integer, primary_key=True) return col_val class Person(TestBase): """single table base class""" - __tablename__ = 'person' + + __tablename__ = "person" class Engineer(Person): """ single table inheritance, no extra cols """ @@ -832,130 +926,143 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): is_(Manager.id.property.columns[0], Person.__table__.c.id) def test_joined_from_single(self): - class Company(Base, fixtures.ComparableEntity): - __tablename__ = 'companies' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - employees = relationship('Person') + __tablename__ = "companies" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + employees = relationship("Person") class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - company_id = Column(Integer, ForeignKey('companies.id')) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + company_id = Column(Integer, ForeignKey("companies.id")) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Manager(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} golf_swing = Column(String(50)) class Engineer(Person): - __tablename__ = 'engineers' - __mapper_args__ = {'polymorphic_identity': 'engineer'} - id = Column(Integer, ForeignKey('people.id'), - primary_key=True) + __tablename__ = "engineers" + __mapper_args__ = {"polymorphic_identity": "engineer"} + id = Column(Integer, ForeignKey("people.id"), primary_key=True) primary_language = Column(String(50)) assert Person.__table__.c.golf_swing is not None - assert 'primary_language' not in Person.__table__.c + assert "primary_language" not in Person.__table__.c assert Engineer.__table__.c.primary_language is not None assert Engineer.primary_language is not None assert Manager.golf_swing is not None - assert not hasattr(Person, 'primary_language') - assert not hasattr(Person, 'golf_swing') - assert not hasattr(Engineer, 'golf_swing') - assert not hasattr(Manager, 'primary_language') + assert not hasattr(Person, "primary_language") + assert not hasattr(Person, "golf_swing") + assert not hasattr(Engineer, "golf_swing") + assert not hasattr(Manager, "primary_language") Base.metadata.create_all() sess = create_session() - e1 = Engineer(name='dilbert', primary_language='java') - e2 = Engineer(name='wally', primary_language='c++') - m1 = Manager(name='dogbert', golf_swing='fore!') - c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1]) - e3 = Engineer(name='vlad', primary_language='cobol') - c2 = Company(name='Elbonia, Inc.', employees=[e3]) + e1 = Engineer(name="dilbert", primary_language="java") + e2 = Engineer(name="wally", primary_language="c++") + m1 = Manager(name="dogbert", golf_swing="fore!") + c1 = Company(name="MegaCorp, Inc.", employees=[e1, e2, m1]) + e3 = Engineer(name="vlad", primary_language="cobol") + c2 = Company(name="Elbonia, Inc.", employees=[e3]) sess.add(c1) sess.add(c2) sess.flush() sess.expunge_all() - eq_(sess.query(Person).with_polymorphic(Engineer). - filter(Engineer.primary_language - == 'cobol').first(), Engineer(name='vlad')) - eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). - any(Engineer.primary_language - == 'cobol')).first(), c2) - eq_(sess.query(Engineer).filter_by(primary_language='cobol' - ).one(), - Engineer(name='vlad', primary_language='cobol')) + eq_( + sess.query(Person) + .with_polymorphic(Engineer) + .filter(Engineer.primary_language == "cobol") + .first(), + Engineer(name="vlad"), + ) + eq_( + sess.query(Company) + .filter( + Company.employees.of_type(Engineer).any( + Engineer.primary_language == "cobol" + ) + ) + .first(), + c2, + ) + eq_( + sess.query(Engineer).filter_by(primary_language="cobol").one(), + Engineer(name="vlad", primary_language="cobol"), + ) def test_single_from_joined_colsonsub(self): class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Manager(Person): - __tablename__ = 'manager' - __mapper_args__ = {'polymorphic_identity': 'manager'} - id = Column(Integer, ForeignKey('people.id'), primary_key=True) + __tablename__ = "manager" + __mapper_args__ = {"polymorphic_identity": "manager"} + id = Column(Integer, ForeignKey("people.id"), primary_key=True) golf_swing = Column(String(50)) class Boss(Manager): boss_name = Column(String(50)) is_( - Boss.__mapper__.column_attrs['boss_name'].columns[0], - Manager.__table__.c.boss_name + Boss.__mapper__.column_attrs["boss_name"].columns[0], + Manager.__table__.c.boss_name, ) def test_polymorphic_on_converted_from_inst(self): class A(Base): - __tablename__ = 'A' + __tablename__ = "A" id = Column(Integer, primary_key=True) discriminator = Column(String) @declared_attr def __mapper_args__(cls): return { - 'polymorphic_identity': cls.__name__, - 'polymorphic_on': cls.discriminator + "polymorphic_identity": cls.__name__, + "polymorphic_on": cls.discriminator, } class B(A): pass + is_(B.__mapper__.polymorphic_on, A.__table__.c.discriminator) def test_add_deferred(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "people" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) Person.name = deferred(Column(String(10))) Base.metadata.create_all() sess = create_session() - p = Person(name='ratbert') + p = Person(name="ratbert") sess.add(p) sess.flush() sess.expunge_all() - eq_(sess.query(Person).all(), [Person(name='ratbert')]) + eq_(sess.query(Person).all(), [Person(name="ratbert")]) sess.expunge_all() - person = sess.query(Person).filter(Person.name == 'ratbert' - ).one() - assert 'name' not in person.__dict__ + person = sess.query(Person).filter(Person.name == "ratbert").one() + assert "name" not in person.__dict__ def test_single_fksonsub(self): """test single inheritance with a foreign key-holding column on @@ -965,78 +1072,90 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} - primary_language_id = Column(Integer, - ForeignKey('languages.id')) - primary_language = relationship('Language') + __mapper_args__ = {"polymorphic_identity": "engineer"} + primary_language_id = Column(Integer, ForeignKey("languages.id")) + primary_language = relationship("Language") class Language(Base, fixtures.ComparableEntity): - __tablename__ = 'languages' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "languages" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) - assert not hasattr(Person, 'primary_language_id') + assert not hasattr(Person, "primary_language_id") Base.metadata.create_all() sess = create_session() - java, cpp, cobol = Language(name='java'), Language(name='cpp'), \ - Language(name='cobol') - e1 = Engineer(name='dilbert', primary_language=java) - e2 = Engineer(name='wally', primary_language=cpp) - e3 = Engineer(name='vlad', primary_language=cobol) + java, cpp, cobol = ( + Language(name="java"), + Language(name="cpp"), + Language(name="cobol"), + ) + e1 = Engineer(name="dilbert", primary_language=java) + e2 = Engineer(name="wally", primary_language=cpp) + e3 = Engineer(name="vlad", primary_language=cobol) sess.add_all([e1, e2, e3]) sess.flush() sess.expunge_all() - eq_(sess.query(Person).filter(Engineer.primary_language.has( - Language.name - == 'cobol')).first(), - Engineer(name='vlad', primary_language=Language(name='cobol'))) - eq_(sess.query(Engineer).filter(Engineer.primary_language.has( - Language.name - == 'cobol')).one(), - Engineer(name='vlad', primary_language=Language(name='cobol'))) - eq_(sess.query(Person).join(Engineer.primary_language).order_by( - Language.name).all(), - [Engineer(name='vlad', - primary_language=Language(name='cobol')), - Engineer(name='wally', primary_language=Language(name='cpp' - )), - Engineer(name='dilbert', primary_language=Language(name='java'))]) + eq_( + sess.query(Person) + .filter(Engineer.primary_language.has(Language.name == "cobol")) + .first(), + Engineer(name="vlad", primary_language=Language(name="cobol")), + ) + eq_( + sess.query(Engineer) + .filter(Engineer.primary_language.has(Language.name == "cobol")) + .one(), + Engineer(name="vlad", primary_language=Language(name="cobol")), + ) + eq_( + sess.query(Person) + .join(Engineer.primary_language) + .order_by(Language.name) + .all(), + [ + Engineer(name="vlad", primary_language=Language(name="cobol")), + Engineer(name="wally", primary_language=Language(name="cpp")), + Engineer( + name="dilbert", primary_language=Language(name="java") + ), + ], + ) def test_single_three_levels(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' + __tablename__ = "people" id = Column(Integer, primary_key=True) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) class JuniorEngineer(Engineer): - __mapper_args__ = \ - {'polymorphic_identity': 'junior_engineer'} + __mapper_args__ = {"polymorphic_identity": "junior_engineer"} nerf_gun = Column(String(50)) class Manager(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} golf_swing = Column(String(50)) assert JuniorEngineer.nerf_gun @@ -1044,33 +1163,31 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): assert JuniorEngineer.name assert Manager.golf_swing assert Engineer.primary_language - assert not hasattr(Engineer, 'golf_swing') - assert not hasattr(Engineer, 'nerf_gun') - assert not hasattr(Manager, 'nerf_gun') - assert not hasattr(Manager, 'primary_language') + assert not hasattr(Engineer, "golf_swing") + assert not hasattr(Engineer, "nerf_gun") + assert not hasattr(Manager, "nerf_gun") + assert not hasattr(Manager, "primary_language") def test_single_detects_conflict(self): - class Person(Base): - __tablename__ = 'people' + __tablename__ = "people" id = Column(Integer, primary_key=True) name = Column(String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} + __mapper_args__ = {"polymorphic_identity": "engineer"} primary_language = Column(String(50)) # test sibling col conflict def go(): - class Manager(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} golf_swing = Column(String(50)) primary_language = Column(String(50)) @@ -1079,72 +1196,64 @@ class DeclarativeInheritanceTest(DeclarativeTestBase): # test parent col conflict def go(): - class Salesman(Person): - __mapper_args__ = {'polymorphic_identity': 'manager'} + __mapper_args__ = {"polymorphic_identity": "manager"} name = Column(String(50)) assert_raises(sa.exc.ArgumentError, go) def test_single_no_special_cols(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} def go(): - class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} - primary_language = Column('primary_language', - String(50)) + __mapper_args__ = {"polymorphic_identity": "engineer"} + primary_language = Column("primary_language", String(50)) foo_bar = Column(Integer, primary_key=True) - assert_raises_message(sa.exc.ArgumentError, - 'place primary key', go) + assert_raises_message(sa.exc.ArgumentError, "place primary key", go) def test_single_no_table_args(self): - class Person(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) - discriminator = Column('type', String(50)) - __mapper_args__ = {'polymorphic_on': discriminator} + __tablename__ = "people" + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) + discriminator = Column("type", String(50)) + __mapper_args__ = {"polymorphic_on": discriminator} def go(): - class Engineer(Person): - __mapper_args__ = {'polymorphic_identity': 'engineer'} - primary_language = Column('primary_language', - String(50)) + __mapper_args__ = {"polymorphic_identity": "engineer"} + primary_language = Column("primary_language", String(50)) # this should be on the Person class, as this is single # table inheritance, which is why we test that this # throws an exception! - __table_args__ = {'mysql_engine': 'InnoDB'} + __table_args__ = {"mysql_engine": "InnoDB"} - assert_raises_message(sa.exc.ArgumentError, - 'place __table_args__', go) + assert_raises_message(sa.exc.ArgumentError, "place __table_args__", go) @testing.emits_warning("This declarative") def test_dupe_name_in_hierarchy(self): class A(Base): __tablename__ = "a" id = Column(Integer, primary_key=True) + a_1 = A class A(a_1): - __tablename__ = 'b' + __tablename__ = "b" id = Column(Integer(), ForeignKey(a_1.id), primary_key=True) assert A.__mapper__.inherits is a_1.__mapper__ @@ -1155,68 +1264,76 @@ class OverlapColPrecedenceTest(DeclarativeTestBase): """test #1892 cases when declarative does column precedence.""" def _run_test(self, Engineer, e_id, p_id): - p_table = Base.metadata.tables['person'] - e_table = Base.metadata.tables['engineer'] + p_table = Base.metadata.tables["person"] + e_table = Base.metadata.tables["engineer"] assert Engineer.id.property.columns[0] is e_table.c[e_id] assert Engineer.id.property.columns[1] is p_table.c[p_id] def test_basic(self): class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column(Integer, primary_key=True) class Engineer(Person): - __tablename__ = 'engineer' - id = Column(Integer, ForeignKey('person.id'), primary_key=True) + __tablename__ = "engineer" + id = Column(Integer, ForeignKey("person.id"), primary_key=True) self._run_test(Engineer, "id", "id") def test_alt_name_base(self): class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column("pid", Integer, primary_key=True) class Engineer(Person): - __tablename__ = 'engineer' - id = Column(Integer, ForeignKey('person.pid'), primary_key=True) + __tablename__ = "engineer" + id = Column(Integer, ForeignKey("person.pid"), primary_key=True) self._run_test(Engineer, "id", "pid") def test_alt_name_sub(self): class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column(Integer, primary_key=True) class Engineer(Person): - __tablename__ = 'engineer' - id = Column("eid", Integer, ForeignKey('person.id'), - primary_key=True) + __tablename__ = "engineer" + id = Column( + "eid", Integer, ForeignKey("person.id"), primary_key=True + ) self._run_test(Engineer, "eid", "id") def test_alt_name_both(self): class Person(Base): - __tablename__ = 'person' + __tablename__ = "person" id = Column("pid", Integer, primary_key=True) class Engineer(Person): - __tablename__ = 'engineer' - id = Column("eid", Integer, ForeignKey('person.pid'), - primary_key=True) + __tablename__ = "engineer" + id = Column( + "eid", Integer, ForeignKey("person.pid"), primary_key=True + ) self._run_test(Engineer, "eid", "pid") class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): - - def _roundtrip(self, Employee, Manager, Engineer, Boss, - polymorphic=True, explicit_type=False): + def _roundtrip( + self, + Employee, + Manager, + Engineer, + Boss, + polymorphic=True, + explicit_type=False, + ): Base.metadata.create_all() sess = create_session() - e1 = Engineer(name='dilbert', primary_language='java') - e2 = Engineer(name='wally', primary_language='c++') - m1 = Manager(name='dogbert', golf_swing='fore!') - e3 = Engineer(name='vlad', primary_language='cobol') + e1 = Engineer(name="dilbert", primary_language="java") + e2 = Engineer(name="wally", primary_language="c++") + m1 = Manager(name="dogbert", golf_swing="fore!") + e3 = Engineer(name="vlad", primary_language="cobol") b1 = Boss(name="pointy haired") if polymorphic: @@ -1228,7 +1345,9 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): AttributeError, "does not implement attribute .?'type' " "at the instance level.", - getattr, obj, "type" + getattr, + obj, + "type", ) else: assert "type" not in Engineer.__dict__ @@ -1239,66 +1358,93 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): sess.flush() sess.expunge_all() if polymorphic: - eq_(sess.query(Employee).order_by(Employee.name).all(), - [Engineer(name='dilbert'), Manager(name='dogbert'), - Boss(name='pointy haired'), - Engineer(name='vlad'), Engineer(name='wally')]) + eq_( + sess.query(Employee).order_by(Employee.name).all(), + [ + Engineer(name="dilbert"), + Manager(name="dogbert"), + Boss(name="pointy haired"), + Engineer(name="vlad"), + Engineer(name="wally"), + ], + ) else: - eq_(sess.query(Engineer).order_by(Engineer.name).all(), - [Engineer(name='dilbert'), Engineer(name='vlad'), - Engineer(name='wally')]) - eq_(sess.query(Manager).all(), [Manager(name='dogbert')]) - eq_(sess.query(Boss).all(), [Boss(name='pointy haired')]) + eq_( + sess.query(Engineer).order_by(Engineer.name).all(), + [ + Engineer(name="dilbert"), + Engineer(name="vlad"), + Engineer(name="wally"), + ], + ) + eq_(sess.query(Manager).all(), [Manager(name="dogbert")]) + eq_(sess.query(Boss).all(), [Boss(name="pointy haired")]) e1 = sess.query(Engineer).order_by(Engineer.name).first() sess.expire(e1) - eq_(e1.name, 'dilbert') + eq_(e1.name, "dilbert") def test_explicit(self): engineers = Table( - 'engineers', Base.metadata, - Column('id', - Integer, primary_key=True, test_needs_autoincrement=True), - Column('name', String(50)), - Column('primary_language', String(50))) - managers = Table('managers', Base.metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('golf_swing', String(50)) - ) - boss = Table('boss', Base.metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(50)), - Column('golf_swing', String(50)) - ) - punion = polymorphic_union({ - 'engineer': engineers, - 'manager': managers, - 'boss': boss}, 'type', 'punion') + "engineers", + Base.metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(50)), + Column("primary_language", String(50)), + ) + managers = Table( + "managers", + Base.metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(50)), + Column("golf_swing", String(50)), + ) + boss = Table( + "boss", + Base.metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(50)), + Column("golf_swing", String(50)), + ) + punion = polymorphic_union( + {"engineer": engineers, "manager": managers, "boss": boss}, + "type", + "punion", + ) class Employee(Base, fixtures.ComparableEntity): __table__ = punion - __mapper_args__ = {'polymorphic_on': punion.c.type} + __mapper_args__ = {"polymorphic_on": punion.c.type} class Engineer(Employee): __table__ = engineers - __mapper_args__ = {'polymorphic_identity': 'engineer', - 'concrete': True} + __mapper_args__ = { + "polymorphic_identity": "engineer", + "concrete": True, + } class Manager(Employee): __table__ = managers - __mapper_args__ = {'polymorphic_identity': 'manager', - 'concrete': True} + __mapper_args__ = { + "polymorphic_identity": "manager", + "concrete": True, + } class Boss(Manager): __table__ = boss - __mapper_args__ = {'polymorphic_identity': 'boss', - 'concrete': True} + __mapper_args__ = { + "polymorphic_identity": "boss", + "concrete": True, + } self._roundtrip(Employee, Manager, Engineer, Boss) @@ -1307,34 +1453,38 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): class Employee(Base, fixtures.ComparableEntity): - __tablename__ = 'people' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "people" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) class Engineer(Employee): - __tablename__ = 'engineers' - __mapper_args__ = {'concrete': True} - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "engineers" + __mapper_args__ = {"concrete": True} + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) primary_language = Column(String(50)) name = Column(String(50)) class Manager(Employee): - __tablename__ = 'manager' - __mapper_args__ = {'concrete': True} - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "manager" + __mapper_args__ = {"concrete": True} + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) golf_swing = Column(String(50)) name = Column(String(50)) class Boss(Manager): - __tablename__ = 'boss' - __mapper_args__ = {'concrete': True} - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "boss" + __mapper_args__ = {"concrete": True} + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) golf_swing = Column(String(50)) name = Column(String(50)) @@ -1345,33 +1495,40 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): pass class Manager(Employee): - __tablename__ = 'manager' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "manager" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) __mapper_args__ = { - 'polymorphic_identity': 'manager', - 'concrete': True} + "polymorphic_identity": "manager", + "concrete": True, + } class Boss(Manager): - __tablename__ = 'boss' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "boss" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) __mapper_args__ = { - 'polymorphic_identity': 'boss', - 'concrete': True} + "polymorphic_identity": "boss", + "concrete": True, + } class Engineer(Employee): - __tablename__ = 'engineer' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "engineer" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) primary_language = Column(String(40)) - __mapper_args__ = {'polymorphic_identity': 'engineer', - 'concrete': True} + __mapper_args__ = { + "polymorphic_identity": "engineer", + "concrete": True, + } self._roundtrip(Employee, Manager, Engineer, Boss) @@ -1382,104 +1539,120 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): return Column(String(50)) class Manager(Employee): - __tablename__ = 'manager' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "manager" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) paperwork = Column(String(10)) __mapper_args__ = { - 'polymorphic_identity': 'manager', 'concrete': True} + "polymorphic_identity": "manager", + "concrete": True, + } class Engineer(Employee): - __tablename__ = 'engineer' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "engineer" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) @property def paperwork(self): return "p" __mapper_args__ = { - 'polymorphic_identity': 'engineer', 'concrete': True} + "polymorphic_identity": "engineer", + "concrete": True, + } Base.metadata.create_all() sess = Session() - sess.add(Engineer(name='d')) + sess.add(Engineer(name="d")) sess.commit() # paperwork is excluded because there's a descritor; so it is # not in the Engineers mapped properties at all, though is inside the # class manager. Maybe it shouldn't be in the class manager either. - assert 'paperwork' in Engineer.__mapper__.class_manager - assert 'paperwork' not in Engineer.__mapper__.attrs.keys() + assert "paperwork" in Engineer.__mapper__.class_manager + assert "paperwork" not in Engineer.__mapper__.attrs.keys() # type currently does get mapped, as a # ConcreteInheritedProperty, which means, "ignore this thing inherited # from the concrete base". if we didn't specify concrete=True, then # this one gets stuck in the error condition also. - assert 'type' in Engineer.__mapper__.class_manager - assert 'type' in Engineer.__mapper__.attrs.keys() + assert "type" in Engineer.__mapper__.class_manager + assert "type" in Engineer.__mapper__.attrs.keys() e1 = sess.query(Engineer).first() - eq_(e1.name, 'd') + eq_(e1.name, "d") sess.expire(e1) - eq_(e1.name, 'd') + eq_(e1.name, "d") def test_concrete_extension(self): class Employee(ConcreteBase, Base, fixtures.ComparableEntity): - __tablename__ = 'employee' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "employee" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) __mapper_args__ = { - 'polymorphic_identity': 'employee', - 'concrete': True} + "polymorphic_identity": "employee", + "concrete": True, + } class Manager(Employee): - __tablename__ = 'manager' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "manager" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) __mapper_args__ = { - 'polymorphic_identity': 'manager', - 'concrete': True} + "polymorphic_identity": "manager", + "concrete": True, + } class Boss(Manager): - __tablename__ = 'boss' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "boss" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) __mapper_args__ = { - 'polymorphic_identity': 'boss', - 'concrete': True} + "polymorphic_identity": "boss", + "concrete": True, + } class Engineer(Employee): - __tablename__ = 'engineer' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "engineer" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) primary_language = Column(String(40)) - __mapper_args__ = {'polymorphic_identity': 'engineer', - 'concrete': True} + __mapper_args__ = { + "polymorphic_identity": "engineer", + "concrete": True, + } + self._roundtrip(Employee, Manager, Engineer, Boss) def test_has_inherited_table_doesnt_consider_base(self): class A(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) assert not has_inherited_table(A) class B(A): - __tablename__ = 'b' - id = Column(Integer, ForeignKey('a.id'), primary_key=True) + __tablename__ = "b" + id = Column(Integer, ForeignKey("a.id"), primary_key=True) assert has_inherited_table(B) def test_has_inherited_table_in_mapper_args(self): class Test(Base): - __tablename__ = 'test' + __tablename__ = "test" id = Column(Integer, primary_key=True) type = Column(String(20)) @@ -1487,15 +1660,15 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): def __mapper_args__(cls): if not has_inherited_table(cls): ret = { - 'polymorphic_identity': 'default', - 'polymorphic_on': cls.type, + "polymorphic_identity": "default", + "polymorphic_on": cls.type, } else: - ret = {'polymorphic_identity': cls.__name__} + ret = {"polymorphic_identity": cls.__name__} return ret class PolyTest(Test): - __tablename__ = 'poly_test' + __tablename__ = "poly_test" id = Column(Integer, ForeignKey(Test.id), primary_key=True) configure_mappers() @@ -1508,9 +1681,10 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): pass class Manager(Employee): - __tablename__ = 'manager' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "manager" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) @@ -1519,13 +1693,15 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): return "manager" __mapper_args__ = { - 'polymorphic_identity': "manager", - 'concrete': True} + "polymorphic_identity": "manager", + "concrete": True, + } class Boss(Manager): - __tablename__ = 'boss' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "boss" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) golf_swing = Column(String(40)) @@ -1534,94 +1710,116 @@ class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): return "boss" __mapper_args__ = { - 'polymorphic_identity': "boss", - 'concrete': True} + "polymorphic_identity": "boss", + "concrete": True, + } class Engineer(Employee): - __tablename__ = 'engineer' - employee_id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "engineer" + employee_id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) primary_language = Column(String(40)) @property def type(self): return "engineer" - __mapper_args__ = {'polymorphic_identity': "engineer", - 'concrete': True} + + __mapper_args__ = { + "polymorphic_identity": "engineer", + "concrete": True, + } + self._roundtrip(Employee, Manager, Engineer, Boss, explicit_type=True) class ConcreteExtensionConfigTest( - _RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase): - __dialect__ = 'default' + _RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase +): + __dialect__ = "default" def test_classreg_setup(self): class A(Base, fixtures.ComparableEntity): - __tablename__ = 'a' - id = Column(Integer, - primary_key=True, test_needs_autoincrement=True) + __tablename__ = "a" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) data = Column(String(50)) - collection = relationship("BC", primaryjoin="BC.a_id == A.id", - collection_class=set) + collection = relationship( + "BC", primaryjoin="BC.a_id == A.id", collection_class=set + ) class BC(AbstractConcreteBase, Base, fixtures.ComparableEntity): pass class B(BC): - __tablename__ = 'b' - id = Column(Integer, - primary_key=True, test_needs_autoincrement=True) + __tablename__ = "b" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) - a_id = Column(Integer, ForeignKey('a.id')) + a_id = Column(Integer, ForeignKey("a.id")) data = Column(String(50)) b_data = Column(String(50)) - __mapper_args__ = { - "polymorphic_identity": "b", - "concrete": True - } + __mapper_args__ = {"polymorphic_identity": "b", "concrete": True} class C(BC): - __tablename__ = 'c' - id = Column(Integer, - primary_key=True, test_needs_autoincrement=True) - a_id = Column(Integer, ForeignKey('a.id')) + __tablename__ = "c" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + a_id = Column(Integer, ForeignKey("a.id")) data = Column(String(50)) c_data = Column(String(50)) - __mapper_args__ = { - "polymorphic_identity": "c", - "concrete": True - } + __mapper_args__ = {"polymorphic_identity": "c", "concrete": True} Base.metadata.create_all() sess = Session() - sess.add_all([ - A(data='a1', collection=set([ - B(data='a1b1', b_data='a1b1'), - C(data='a1b2', c_data='a1c1'), - B(data='a1b2', b_data='a1b2'), - C(data='a1c2', c_data='a1c2'), - ])), - A(data='a2', collection=set([ - B(data='a2b1', b_data='a2b1'), - C(data='a2c1', c_data='a2c1'), - B(data='a2b2', b_data='a2b2'), - C(data='a2c2', c_data='a2c2'), - ])) - ]) + sess.add_all( + [ + A( + data="a1", + collection=set( + [ + B(data="a1b1", b_data="a1b1"), + C(data="a1b2", c_data="a1c1"), + B(data="a1b2", b_data="a1b2"), + C(data="a1c2", c_data="a1c2"), + ] + ), + ), + A( + data="a2", + collection=set( + [ + B(data="a2b1", b_data="a2b1"), + C(data="a2c1", c_data="a2c1"), + B(data="a2b2", b_data="a2b2"), + C(data="a2c2", c_data="a2c2"), + ] + ), + ), + ] + ) sess.commit() sess.expunge_all() eq_( - sess.query(A).filter_by(data='a2').all(), + sess.query(A).filter_by(data="a2").all(), [ - A(data='a2', collection=set([ - B(data='a2b1', b_data='a2b1'), - B(data='a2b2', b_data='a2b2'), - C(data='a2c1', c_data='a2c1'), - C(data='a2c2', c_data='a2c2'), - ])) - ] + A( + data="a2", + collection=set( + [ + B(data="a2b1", b_data="a2b1"), + B(data="a2b2", b_data="a2b2"), + C(data="a2c1", c_data="a2c1"), + C(data="a2c2", c_data="a2c2"), + ] + ), + ) + ], ) self.assert_compile( @@ -1632,7 +1830,7 @@ class ConcreteExtensionConfigTest( "'c' AS type FROM c UNION ALL SELECT b.id AS id, b.a_id AS a_id, " "b.data AS data, CAST(NULL AS VARCHAR(50)) AS c_data, " "b.b_data AS b_data, 'b' AS type FROM b) AS pjoin " - "ON pjoin.a_id = a.id" + "ON pjoin.a_id = a.id", ) def test_prop_on_base(self): @@ -1641,7 +1839,7 @@ class ConcreteExtensionConfigTest( counter = mock.Mock() class Something(Base): - __tablename__ = 'something' + __tablename__ = "something" id = Column(Integer, primary_key=True) class AbstractConcreteAbstraction(AbstractConcreteBase, Base): @@ -1664,23 +1862,22 @@ class ConcreteExtensionConfigTest( return relationship("Something") class ConcreteConcreteAbstraction(AbstractConcreteAbstraction): - __tablename__ = 'cca' - __mapper_args__ = { - 'polymorphic_identity': 'ccb', - 'concrete': True} + __tablename__ = "cca" + __mapper_args__ = {"polymorphic_identity": "ccb", "concrete": True} # concrete is mapped, the abstract base is not (yet) assert ConcreteConcreteAbstraction.__mapper__ - assert not hasattr(AbstractConcreteAbstraction, '__mapper__') + assert not hasattr(AbstractConcreteAbstraction, "__mapper__") session = Session() self.assert_compile( session.query(ConcreteConcreteAbstraction).filter( - ConcreteConcreteAbstraction.something.has(id=1)), + ConcreteConcreteAbstraction.something.has(id=1) + ), "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, " "cca.something_id AS cca_something_id FROM cca WHERE EXISTS " "(SELECT 1 FROM something WHERE something.id = cca.something_id " - "AND something.id = :id_1)" + "AND something.id = :id_1)", ) # now it is @@ -1688,35 +1885,38 @@ class ConcreteExtensionConfigTest( self.assert_compile( session.query(ConcreteConcreteAbstraction).filter( - ConcreteConcreteAbstraction.something_else.has(id=1)), + ConcreteConcreteAbstraction.something_else.has(id=1) + ), "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, " "cca.something_id AS cca_something_id FROM cca WHERE EXISTS " "(SELECT 1 FROM something WHERE something.id = cca.something_id " - "AND something.id = :id_1)" + "AND something.id = :id_1)", ) self.assert_compile( session.query(AbstractConcreteAbstraction).filter( - AbstractConcreteAbstraction.something.has(id=1)), + AbstractConcreteAbstraction.something.has(id=1) + ), "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, " "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, " "pjoin.type AS pjoin_type FROM " "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, " "cca.something_id AS something_id, 'ccb' AS type FROM cca) " "AS pjoin WHERE EXISTS (SELECT 1 FROM something " - "WHERE something.id = pjoin.something_id AND something.id = :id_1)" + "WHERE something.id = pjoin.something_id AND something.id = :id_1)", ) self.assert_compile( session.query(AbstractConcreteAbstraction).filter( - AbstractConcreteAbstraction.something_else.has(id=1)), + AbstractConcreteAbstraction.something_else.has(id=1) + ), "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, " "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, " "pjoin.type AS pjoin_type FROM " "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, " "cca.something_id AS something_id, 'ccb' AS type FROM cca) " "AS pjoin WHERE EXISTS (SELECT 1 FROM something " - "WHERE something.id = pjoin.something_id AND something.id = :id_1)" + "WHERE something.id = pjoin.something_id AND something.id = :id_1)", ) def test_abstract_in_hierarchy(self): @@ -1729,10 +1929,11 @@ class ConcreteExtensionConfigTest( send_method = Column(String) class ActualDocument(ContactDocument): - __tablename__ = 'actual_documents' + __tablename__ = "actual_documents" __mapper_args__ = { - 'concrete': True, - 'polymorphic_identity': 'actual'} + "concrete": True, + "polymorphic_identity": "actual", + } id = Column(Integer, primary_key=True) @@ -1746,22 +1947,20 @@ class ConcreteExtensionConfigTest( "FROM (SELECT actual_documents.doctype AS doctype, " "actual_documents.send_method AS send_method, " "actual_documents.id AS id, 'actual' AS type " - "FROM actual_documents) AS pjoin" + "FROM actual_documents) AS pjoin", ) def test_column_attr_names(self): """test #3480""" class Document(Base, AbstractConcreteBase): - documentType = Column('documenttype', String) + documentType = Column("documenttype", String) class Offer(Document): - __tablename__ = 'offers' + __tablename__ = "offers" id = Column(Integer, primary_key=True) - __mapper_args__ = { - 'polymorphic_identity': 'offer' - } + __mapper_args__ = {"polymorphic_identity": "offer"} configure_mappers() session = Session() @@ -1770,12 +1969,12 @@ class ConcreteExtensionConfigTest( "SELECT pjoin.documenttype AS pjoin_documenttype, " "pjoin.id AS pjoin_id, pjoin.type AS pjoin_type FROM " "(SELECT offers.documenttype AS documenttype, offers.id AS id, " - "'offer' AS type FROM offers) AS pjoin" + "'offer' AS type FROM offers) AS pjoin", ) self.assert_compile( session.query(Document.documentType), "SELECT pjoin.documenttype AS pjoin_documenttype FROM " "(SELECT offers.documenttype AS documenttype, offers.id AS id, " - "'offer' AS type FROM offers) AS pjoin" + "'offer' AS type FROM offers) AS pjoin", ) |