diff options
Diffstat (limited to 'test/orm/inheritance/test_basic.py')
| -rw-r--r-- | test/orm/inheritance/test_basic.py | 2219 |
1 files changed, 1297 insertions, 922 deletions
diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py index 7fd9329f9..d871a6a98 100644 --- a/test/orm/inheritance/test_basic.py +++ b/test/orm/inheritance/test_basic.py @@ -19,25 +19,33 @@ from sqlalchemy.testing.util import gc_collect class O2MTest(fixtures.MappedTest): """deals with inheritance and one-to-many relationships""" + @classmethod def define_tables(cls, metadata): global foo, bar, blub - foo = Table('foo', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(20))) - - bar = Table('bar', metadata, - Column('id', Integer, ForeignKey( - 'foo.id'), primary_key=True), - Column('bar_data', String(20))) - - blub = Table('blub', metadata, - Column('id', Integer, ForeignKey( - 'bar.id'), primary_key=True), - Column('foo_id', Integer, ForeignKey( - 'foo.id'), nullable=False), - Column('blub_data', String(20))) + foo = Table( + "foo", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(20)), + ) + + bar = Table( + "bar", + metadata, + Column("id", Integer, ForeignKey("foo.id"), primary_key=True), + Column("bar_data", String(20)), + ) + + blub = Table( + "blub", + metadata, + Column("id", Integer, ForeignKey("bar.id"), primary_key=True), + Column("foo_id", Integer, ForeignKey("foo.id"), nullable=False), + Column("blub_data", String(20)), + ) def test_basic(self): class Foo(object): @@ -46,6 +54,7 @@ class O2MTest(fixtures.MappedTest): def __repr__(self): return "Foo id %d, data %s" % (self.id, self.data) + mapper(Foo, foo) class Bar(Foo): @@ -58,9 +67,12 @@ class O2MTest(fixtures.MappedTest): def __repr__(self): return "Blub id %d, data %s" % (self.id, self.data) - mapper(Blub, blub, inherits=Bar, properties={ - 'parent_foo': relationship(Foo) - }) + mapper( + Blub, + blub, + inherits=Bar, + properties={"parent_foo": relationship(Foo)}, + ) sess = create_session() b1 = Blub("blub #1") @@ -72,59 +84,64 @@ class O2MTest(fixtures.MappedTest): b1.parent_foo = f b2.parent_foo = f sess.flush() - compare = ','.join([repr(b1), repr(b2), repr(b1.parent_foo), - repr(b2.parent_foo)]) + compare = ",".join( + [repr(b1), repr(b2), repr(b1.parent_foo), repr(b2.parent_foo)] + ) sess.expunge_all() result = sess.query(Blub).all() - result_str = ','.join([repr(result[0]), repr(result[1]), - repr(result[0].parent_foo), - repr(result[1].parent_foo)]) + result_str = ",".join( + [ + repr(result[0]), + repr(result[1]), + repr(result[0].parent_foo), + repr(result[1].parent_foo), + ] + ) eq_(compare, result_str) - eq_(result[0].parent_foo.data, 'foo #1') - eq_(result[1].parent_foo.data, 'foo #1') + eq_(result[0].parent_foo.data, "foo #1") + eq_(result[1].parent_foo.data, "foo #1") class PolyExpressionEagerLoad(fixtures.DeclarativeMappedTest): - run_setup_mappers = 'once' - __dialect__ = 'default' + run_setup_mappers = "once" + __dialect__ = "default" @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic class A(fixtures.ComparableEntity, Base): - __tablename__ = 'a' + __tablename__ = "a" - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) discriminator = Column(String(50), nullable=False) - child_id = Column(Integer, ForeignKey('a.id')) - child = relationship('A') + child_id = Column(Integer, ForeignKey("a.id")) + child = relationship("A") - p_a = case([ - (discriminator == "a", "a"), - ], else_="b") + p_a = case([(discriminator == "a", "a")], else_="b") __mapper_args__ = { - 'polymorphic_identity': 'a', + "polymorphic_identity": "a", "polymorphic_on": p_a, } class B(A): - __mapper_args__ = { - 'polymorphic_identity': 'b' - } + __mapper_args__ = {"polymorphic_identity": "b"} @classmethod def insert_data(cls): A = cls.classes.A session = Session(testing.db) - session.add_all([ - A(id=1, discriminator='a'), - A(id=2, discriminator='b', child_id=1), - A(id=3, discriminator='c', child_id=1), - ]) + session.add_all( + [ + A(id=1, discriminator="a"), + A(id=2, discriminator="b", child_id=1), + A(id=3, discriminator="c", child_id=1), + ] + ) session.commit() def test_joinedload(self): @@ -132,46 +149,49 @@ class PolyExpressionEagerLoad(fixtures.DeclarativeMappedTest): B = self.classes.B session = Session(testing.db) - result = session.query(A).filter_by(child_id=None).\ - options(joinedload('child')).one() - - eq_( - result, - A(id=1, discriminator='a', child=[B(id=2), B(id=3)]), + result = ( + session.query(A) + .filter_by(child_id=None) + .options(joinedload("child")) + .one() ) + eq_(result, A(id=1, discriminator="a", child=[B(id=2), B(id=3)])) + -class PolymorphicResolutionMultiLevel(fixtures.DeclarativeMappedTest, - testing.AssertsCompiledSQL): - run_setup_mappers = 'once' - __dialect__ = 'default' +class PolymorphicResolutionMultiLevel( + fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL +): + run_setup_mappers = "once" + __dialect__ = "default" @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic class A(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) 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) class C(A): - __tablename__ = 'c' - id = Column(Integer, ForeignKey('a.id'), primary_key=True) + __tablename__ = "c" + id = Column(Integer, ForeignKey("a.id"), primary_key=True) class D(B): - __tablename__ = 'd' - id = Column(Integer, ForeignKey('b.id'), primary_key=True) + __tablename__ = "d" + id = Column(Integer, ForeignKey("b.id"), primary_key=True) def test_ordered_b_d(self): a_mapper = inspect(self.classes.A) eq_( a_mapper._mappers_from_spec( - [self.classes.B, self.classes.D], None), - [a_mapper, inspect(self.classes.B), inspect(self.classes.D)] + [self.classes.B, self.classes.D], None + ), + [a_mapper, inspect(self.classes.B), inspect(self.classes.D)], ) def test_a(self): @@ -183,10 +203,9 @@ class PolymorphicResolutionMultiLevel(fixtures.DeclarativeMappedTest, spec = [self.classes.D, self.classes.B] eq_( a_mapper._mappers_from_spec( - spec, - self.classes.B.__table__.join(self.classes.D.__table__) + spec, self.classes.B.__table__.join(self.classes.D.__table__) ), - [inspect(self.classes.B), inspect(self.classes.D)] + [inspect(self.classes.B), inspect(self.classes.D)], ) def test_d_selectable(self): @@ -194,10 +213,9 @@ class PolymorphicResolutionMultiLevel(fixtures.DeclarativeMappedTest, spec = [self.classes.D] eq_( a_mapper._mappers_from_spec( - spec, - self.classes.B.__table__.join(self.classes.D.__table__) + spec, self.classes.B.__table__.join(self.classes.D.__table__) ), - [inspect(self.classes.D)] + [inspect(self.classes.D)], ) def test_reverse_d_b(self): @@ -205,52 +223,63 @@ class PolymorphicResolutionMultiLevel(fixtures.DeclarativeMappedTest, spec = [self.classes.D, self.classes.B] eq_( a_mapper._mappers_from_spec(spec, None), - [a_mapper, inspect(self.classes.B), inspect(self.classes.D)] + [a_mapper, inspect(self.classes.B), inspect(self.classes.D)], ) mappers, selectable = a_mapper._with_polymorphic_args(spec=spec) - self.assert_compile(selectable, - "a LEFT OUTER JOIN b ON a.id = b.id " - "LEFT OUTER JOIN d ON b.id = d.id") + self.assert_compile( + selectable, + "a LEFT OUTER JOIN b ON a.id = b.id " + "LEFT OUTER JOIN d ON b.id = d.id", + ) def test_d_b_missing(self): a_mapper = inspect(self.classes.A) spec = [self.classes.D] eq_( a_mapper._mappers_from_spec(spec, None), - [a_mapper, inspect(self.classes.B), inspect(self.classes.D)] + [a_mapper, inspect(self.classes.B), inspect(self.classes.D)], ) mappers, selectable = a_mapper._with_polymorphic_args(spec=spec) - self.assert_compile(selectable, - "a LEFT OUTER JOIN b ON a.id = b.id " - "LEFT OUTER JOIN d ON b.id = d.id") + self.assert_compile( + selectable, + "a LEFT OUTER JOIN b ON a.id = b.id " + "LEFT OUTER JOIN d ON b.id = d.id", + ) def test_d_c_b(self): a_mapper = inspect(self.classes.A) spec = [self.classes.D, self.classes.C, self.classes.B] ms = a_mapper._mappers_from_spec(spec, None) - eq_( - ms[-1], inspect(self.classes.D) - ) + eq_(ms[-1], inspect(self.classes.D)) eq_(ms[0], a_mapper) - eq_( - set(ms[1:3]), set(a_mapper._inheriting_mappers) - ) + eq_(set(ms[1:3]), set(a_mapper._inheriting_mappers)) class PolymorphicOnNotLocalTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - t1 = Table('t1', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('x', String(10)), - Column('q', String(10))) - t2 = Table('t2', metadata, - Column('t2id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('y', String(10)), - Column('xid', ForeignKey('t1.id'))) + t1 = Table( + "t1", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("x", String(10)), + Column("q", String(10)), + ) + t2 = Table( + "t2", + metadata, + Column( + "t2id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("y", String(10)), + Column("xid", ForeignKey("t1.id")), + ) @classmethod def setup_classes(cls): @@ -269,7 +298,9 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): "value 'im not a column' - no " "attribute is mapped to this name.", mapper, - Parent, t2, polymorphic_on="im not a column" + Parent, + t2, + polymorphic_on="im not a column", ) def test_polymorphic_on_non_expr_prop(self): @@ -279,15 +310,15 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): t1t2_join = select([t1.c.x], from_obj=[t1.join(t2)]).alias() def go(): - interface_m = mapper(Parent, t2, - polymorphic_on=lambda: "hi", - polymorphic_identity=0) + interface_m = mapper( + Parent, t2, polymorphic_on=lambda: "hi", polymorphic_identity=0 + ) assert_raises_message( sa_exc.ArgumentError, "Only direct column-mapped property or " "SQL expression can be passed for polymorphic_on", - go + go, ) def test_polymorphic_on_not_present_col(self): @@ -297,15 +328,19 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): def go(): t1t2_join_2 = select([t1.c.q], from_obj=[t1.join(t2)]).alias() - interface_m = mapper(Parent, t2, - polymorphic_on=t1t2_join.c.x, - with_polymorphic=('*', t1t2_join_2), - polymorphic_identity=0) + interface_m = mapper( + Parent, + t2, + polymorphic_on=t1t2_join.c.x, + with_polymorphic=("*", t1t2_join_2), + polymorphic_identity=0, + ) + assert_raises_message( sa_exc.InvalidRequestError, "Could not map polymorphic_on column 'x' to the mapped table - " "polymorphic loads will not function properly", - go + go, ) def test_polymorphic_on_only_in_with_poly(self): @@ -313,10 +348,13 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): Parent = self.classes.Parent t1t2_join = select([t1.c.x], from_obj=[t1.join(t2)]).alias() # if its in the with_polymorphic, then its OK - mapper(Parent, t2, - polymorphic_on=t1t2_join.c.x, - with_polymorphic=('*', t1t2_join), - polymorphic_identity=0) + mapper( + Parent, + t2, + polymorphic_on=t1t2_join.c.x, + with_polymorphic=("*", t1t2_join), + polymorphic_identity=0, + ) def test_polymorpic_on_not_in_with_poly(self): t2, t1 = self.tables.t2, self.tables.t1 @@ -327,123 +365,112 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): # if with_polymorphic, but its not present, not OK def go(): t1t2_join_2 = select([t1.c.q], from_obj=[t1.join(t2)]).alias() - interface_m = mapper(Parent, t2, - polymorphic_on=t1t2_join.c.x, - with_polymorphic=('*', t1t2_join_2), - polymorphic_identity=0) + interface_m = mapper( + Parent, + t2, + polymorphic_on=t1t2_join.c.x, + with_polymorphic=("*", t1t2_join_2), + polymorphic_identity=0, + ) + assert_raises_message( sa_exc.InvalidRequestError, "Could not map polymorphic_on column 'x' " "to the mapped table - " "polymorphic loads will not function properly", - go + go, ) def test_polymorphic_on_expr_explicit_map(self): t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]) - mapper(Parent, t1, properties={ - "discriminator": column_property(expr) - }, polymorphic_identity="parent", - polymorphic_on=expr) - mapper(Child, t2, inherits=Parent, - polymorphic_identity="child") - - self._roundtrip(parent_ident='p', child_ident='c') + expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]) + mapper( + Parent, + t1, + properties={"discriminator": column_property(expr)}, + polymorphic_identity="parent", + polymorphic_on=expr, + ) + mapper(Child, t2, inherits=Parent, polymorphic_identity="child") + + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_expr_implicit_map_no_label_joined(self): t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]) - mapper(Parent, t1, polymorphic_identity="parent", - polymorphic_on=expr) + expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]) + mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr) mapper(Child, t2, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_expr_implicit_map_w_label_joined(self): t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]).label(None) - mapper(Parent, t1, polymorphic_identity="parent", - polymorphic_on=expr) + expr = case( + [(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")] + ).label(None) + mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr) mapper(Child, t2, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_expr_implicit_map_no_label_single(self): """test that single_table_criterion is propagated with a standalone expr""" t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]) - mapper(Parent, t1, polymorphic_identity="parent", - polymorphic_on=expr) + expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]) + mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr) mapper(Child, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_expr_implicit_map_w_label_single(self): """test that single_table_criterion is propagated with a standalone expr""" t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]).label(None) - mapper(Parent, t1, polymorphic_identity="parent", - polymorphic_on=expr) + expr = case( + [(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")] + ).label(None) + mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr) mapper(Child, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_column_prop(self): t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]) + expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]) cprop = column_property(expr) - mapper(Parent, t1, properties={ - "discriminator": cprop - }, polymorphic_identity="parent", - polymorphic_on=cprop) - mapper(Child, t2, inherits=Parent, - polymorphic_identity="child") + mapper( + Parent, + t1, + properties={"discriminator": cprop}, + polymorphic_identity="parent", + polymorphic_on=cprop, + ) + mapper(Child, t2, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_column_str_prop(self): t2, t1 = self.tables.t2, self.tables.t1 Parent, Child = self.classes.Parent, self.classes.Child - expr = case([ - (t1.c.x == "p", "parent"), - (t1.c.x == "c", "child"), - ]) + expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]) cprop = column_property(expr) - mapper(Parent, t1, properties={ - "discriminator": cprop - }, polymorphic_identity="parent", - polymorphic_on="discriminator") - mapper(Child, t2, inherits=Parent, - polymorphic_identity="child") + mapper( + Parent, + t1, + properties={"discriminator": cprop}, + polymorphic_identity="parent", + polymorphic_on="discriminator", + ) + mapper(Child, t2, inherits=Parent, polymorphic_identity="child") - self._roundtrip(parent_ident='p', child_ident='c') + self._roundtrip(parent_ident="p", child_ident="c") def test_polymorphic_on_synonym(self): t2, t1 = self.tables.t2, self.tables.t1 @@ -453,14 +480,17 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): sa_exc.ArgumentError, "Only direct column-mapped property or " "SQL expression can be passed for polymorphic_on", - mapper, Parent, t1, properties={ - "discriminator": cprop, - "discrim_syn": synonym(cprop) - }, polymorphic_identity="parent", - polymorphic_on="discrim_syn") - - def _roundtrip(self, set_event=True, parent_ident='parent', - child_ident='child'): + mapper, + Parent, + t1, + properties={"discriminator": cprop, "discrim_syn": synonym(cprop)}, + polymorphic_identity="parent", + polymorphic_on="discrim_syn", + ) + + def _roundtrip( + self, set_event=True, parent_ident="parent", child_ident="child" + ): Parent, Child = self.classes.Parent, self.classes.Child # locate the "polymorphic_on" ColumnProperty. This isn't @@ -471,55 +501,57 @@ class PolymorphicOnNotLocalTest(fixtures.MappedTest): break else: prop = parent_mapper._columntoproperty[ - parent_mapper.polymorphic_on] + parent_mapper.polymorphic_on + ] # then make sure the column we will query on matches. - is_( - parent_mapper.polymorphic_on, - prop.columns[0] - ) + is_(parent_mapper.polymorphic_on, prop.columns[0]) if set_event: + @event.listens_for(Parent, "init", propagate=True) def set_identity(instance, *arg, **kw): ident = object_mapper(instance).polymorphic_identity - if ident == 'parent': + if ident == "parent": instance.x = parent_ident - elif ident == 'child': + elif ident == "child": instance.x = child_ident else: assert False, "Got unexpected identity %r" % ident s = Session(testing.db) - s.add_all([ - Parent(q="p1"), - Child(q="c1", y="c1"), - Parent(q="p2"), - ]) + s.add_all([Parent(q="p1"), Child(q="c1", y="c1"), Parent(q="p2")]) s.commit() s.close() eq_( [type(t) for t in s.query(Parent).order_by(Parent.id)], - [Parent, Child, Parent] + [Parent, Child, Parent], ) - eq_( - [type(t) for t in s.query(Child).all()], - [Child] - ) + eq_([type(t) for t in s.query(Child).all()], [Child]) class SortOnlyOnImportantFKsTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('a', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('b_id', Integer, - ForeignKey('b.id', use_alter=True, name='b_fk'))) - Table('b', metadata, - Column('id', Integer, ForeignKey('a.id'), primary_key=True)) + Table( + "a", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column( + "b_id", + Integer, + ForeignKey("b.id", use_alter=True, name="b_fk"), + ), + ) + Table( + "b", + metadata, + Column("id", Integer, ForeignKey("a.id"), primary_key=True), + ) @classmethod def setup_classes(cls): @@ -528,16 +560,17 @@ class SortOnlyOnImportantFKsTest(fixtures.MappedTest): class A(Base): __tablename__ = "a" - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - b_id = Column(Integer, ForeignKey('b.id')) + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + b_id = Column(Integer, ForeignKey("b.id")) class B(A): __tablename__ = "b" - id = Column(Integer, ForeignKey('a.id'), primary_key=True) + id = Column(Integer, ForeignKey("a.id"), primary_key=True) - __mapper_args__ = {'inherit_condition': id == A.id} + __mapper_args__ = {"inherit_condition": id == A.id} cls.classes.A = A cls.classes.B = B @@ -552,10 +585,14 @@ class FalseDiscriminatorTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global t1 - t1 = Table('t1', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', Boolean, nullable=False)) + t1 = Table( + "t1", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", Boolean, nullable=False), + ) def test_false_on_sub(self): class Foo(object): @@ -563,6 +600,7 @@ class FalseDiscriminatorTest(fixtures.MappedTest): class Bar(Foo): pass + mapper(Foo, t1, polymorphic_on=t1.c.type, polymorphic_identity=True) mapper(Bar, inherits=Foo, polymorphic_identity=False) sess = create_session() @@ -579,6 +617,7 @@ class FalseDiscriminatorTest(fixtures.MappedTest): class Bat(Ding): pass + mapper(Ding, t1, polymorphic_on=t1.c.type, polymorphic_identity=False) mapper(Bat, inherits=Ding, polymorphic_identity=True) sess = create_session() @@ -594,15 +633,21 @@ class PolymorphicSynonymTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global t1, t2 - t1 = Table('t1', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(10), nullable=False), - Column('info', String(255))) - t2 = Table('t2', metadata, - Column('id', Integer, ForeignKey('t1.id'), - primary_key=True), - Column('data', String(10), nullable=False)) + t1 = Table( + "t1", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(10), nullable=False), + Column("info", String(255)), + ) + t2 = Table( + "t2", + metadata, + Column("id", Integer, ForeignKey("t1.id"), primary_key=True), + Column("data", String(10), nullable=False), + ) def test_polymorphic_synonym(self): class T1(fixtures.ComparableEntity): @@ -611,50 +656,66 @@ class PolymorphicSynonymTest(fixtures.MappedTest): def _set_info(self, x): self._info = x + info = property(info, _set_info) class T2(T1): pass - mapper(T1, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1', - properties={'info': synonym('_info', map_column=True)}) - mapper(T2, t2, inherits=T1, polymorphic_identity='t2') + mapper( + T1, + t1, + polymorphic_on=t1.c.type, + polymorphic_identity="t1", + properties={"info": synonym("_info", map_column=True)}, + ) + mapper(T2, t2, inherits=T1, polymorphic_identity="t2") sess = create_session() - at1 = T1(info='at1') - at2 = T2(info='at2', data='t2 data') + at1 = T1(info="at1") + at2 = T2(info="at2", data="t2 data") sess.add(at1) sess.add(at2) sess.flush() sess.expunge_all() - eq_(sess.query(T2).filter(T2.info == 'at2').one(), at2) + eq_(sess.query(T2).filter(T2.info == "at2").one(), at2) eq_(at2.info, "THE INFO IS:at2") class PolymorphicAttributeManagementTest(fixtures.MappedTest): """Test polymorphic_on can be assigned, can be mirrored, etc.""" - run_setup_mappers = 'once' + run_setup_mappers = "once" @classmethod def define_tables(cls, metadata): - Table('table_a', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('class_name', String(50))) - Table('table_b', metadata, - Column('id', Integer, ForeignKey('table_a.id'), - primary_key=True), - Column('class_name', String(50))) - Table('table_c', metadata, - Column('id', Integer, ForeignKey('table_b.id'), - primary_key=True), - Column('data', String(10))) + Table( + "table_a", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("class_name", String(50)), + ) + Table( + "table_b", + metadata, + Column("id", Integer, ForeignKey("table_a.id"), primary_key=True), + Column("class_name", String(50)), + ) + Table( + "table_c", + metadata, + Column("id", Integer, ForeignKey("table_b.id"), primary_key=True), + Column("data", String(10)), + ) @classmethod def setup_classes(cls): - table_b, table_c, table_a = (cls.tables.table_b, - cls.tables.table_c, - cls.tables.table_a) + table_b, table_c, table_a = ( + cls.tables.table_b, + cls.tables.table_c, + cls.tables.table_a, + ) class A(cls.Basic): pass @@ -668,15 +729,24 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): class D(B): pass - mapper(A, table_a, - polymorphic_on=table_a.c.class_name, - polymorphic_identity='a') - mapper(B, table_b, inherits=A, polymorphic_on=table_b.c.class_name, - polymorphic_identity='b', - properties=dict( - class_name=[table_a.c.class_name, table_b.c.class_name])) - mapper(C, table_c, inherits=B, polymorphic_identity='c') - mapper(D, inherits=B, polymorphic_identity='d') + mapper( + A, + table_a, + polymorphic_on=table_a.c.class_name, + polymorphic_identity="a", + ) + mapper( + B, + table_b, + inherits=A, + polymorphic_on=table_b.c.class_name, + polymorphic_identity="b", + properties=dict( + class_name=[table_a.c.class_name, table_b.c.class_name] + ), + ) + mapper(C, table_c, inherits=B, polymorphic_identity="c") + mapper(D, inherits=B, polymorphic_identity="d") def test_poly_configured_immediate(self): A, C, B = (self.classes.A, self.classes.C, self.classes.B) @@ -684,9 +754,9 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): a = A() b = B() c = C() - eq_(a.class_name, 'a') - eq_(b.class_name, 'b') - eq_(c.class_name, 'c') + eq_(a.class_name, "a") + eq_(b.class_name, "b") + eq_(c.class_name, "c") def test_base_class(self): A, C, B = (self.classes.A, self.classes.C, self.classes.B) @@ -710,7 +780,7 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess = Session() b1 = B() - b1.class_name = 'd' + b1.class_name = "d" sess.add(b1) sess.commit() sess.close() @@ -724,14 +794,14 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess = Session() c1 = C() - c1.class_name = 'b' + c1.class_name = "b" sess.add(c1) assert_raises_message( sa_exc.SAWarning, "Flushing object %s with incompatible " "polymorphic identity 'b'; the object may not " "refresh and/or load correctly" % instance_str(c1), - sess.flush + sess.flush, ) def test_invalid_assignment_upwards(self): @@ -743,14 +813,14 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess = Session() b1 = B() - b1.class_name = 'c' + b1.class_name = "c" sess.add(b1) assert_raises_message( sa_exc.SAWarning, "Flushing object %s with incompatible " "polymorphic identity 'c'; the object may not " "refresh and/or load correctly" % instance_str(b1), - sess.flush + sess.flush, ) def test_entirely_oob_assignment(self): @@ -760,14 +830,14 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess = Session() b1 = B() - b1.class_name = 'xyz' + b1.class_name = "xyz" sess.add(b1) assert_raises_message( sa_exc.SAWarning, "Flushing object %s with incompatible " "polymorphic identity 'xyz'; the object may not " "refresh and/or load correctly" % instance_str(b1), - sess.flush + sess.flush, ) def test_not_set_on_upate(self): @@ -779,7 +849,7 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess.commit() sess.expire(c1) - c1.data = 'foo' + c1.data = "foo" sess.flush() def test_validate_on_upate(self): @@ -791,13 +861,13 @@ class PolymorphicAttributeManagementTest(fixtures.MappedTest): sess.commit() sess.expire(c1) - c1.class_name = 'b' + c1.class_name = "b" assert_raises_message( sa_exc.SAWarning, "Flushing object %s with incompatible " "polymorphic identity 'b'; the object may not " "refresh and/or load correctly" % instance_str(c1), - sess.flush + sess.flush, ) @@ -809,27 +879,41 @@ class CascadeTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global t1, t2, t3, t4 - t1 = Table('t1', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(30))) - - t2 = Table('t2', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('t1id', Integer, ForeignKey('t1.id')), - Column('type', String(30)), - Column('data', String(30))) - t3 = Table('t3', metadata, - Column('id', Integer, ForeignKey('t2.id'), - primary_key=True), - Column('moredata', String(30))) - - t4 = Table('t4', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('t3id', Integer, ForeignKey('t3.id')), - Column('data', String(30))) + t1 = Table( + "t1", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(30)), + ) + + t2 = Table( + "t2", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("t1id", Integer, ForeignKey("t1.id")), + Column("type", String(30)), + Column("data", String(30)), + ) + t3 = Table( + "t3", + metadata, + Column("id", Integer, ForeignKey("t2.id"), primary_key=True), + Column("moredata", String(30)), + ) + + t4 = Table( + "t4", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("t3id", Integer, ForeignKey("t3.id")), + Column("data", String(30)), + ) def test_cascade(self): class T1(fixtures.BasicEntity): @@ -844,25 +928,27 @@ class CascadeTest(fixtures.MappedTest): class T4(fixtures.BasicEntity): pass - mapper(T1, t1, properties={ - 't2s': relationship(T2, cascade="all") - }) - mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2') - mapper(T3, t3, inherits=T2, polymorphic_identity='t3', properties={ - 't4s': relationship(T4, cascade="all") - }) + mapper(T1, t1, properties={"t2s": relationship(T2, cascade="all")}) + mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity="t2") + mapper( + T3, + t3, + inherits=T2, + polymorphic_identity="t3", + properties={"t4s": relationship(T4, cascade="all")}, + ) mapper(T4, t4) sess = create_session() - t1_1 = T1(data='t1') + t1_1 = T1(data="t1") - t3_1 = T3(data='t3', moredata='t3') - t2_1 = T2(data='t2') + t3_1 = T3(data="t3", moredata="t3") + t2_1 = T2(data="t2") t1_1.t2s.append(t2_1) t1_1.t2s.append(t3_1) - t4_1 = T4(data='t4') + t4_1 = T4(data="t4") t3_1.t4s.append(t4_1) sess.add(t1_1) @@ -878,21 +964,34 @@ class CascadeTest(fixtures.MappedTest): class M2OUseGetTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('base', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(30))) - Table('sub', metadata, - Column('id', Integer, ForeignKey('base.id'), primary_key=True)) - Table('related', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('sub_id', Integer, ForeignKey('sub.id'))) + Table( + "base", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(30)), + ) + Table( + "sub", + metadata, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + ) + Table( + "related", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("sub_id", Integer, ForeignKey("sub.id")), + ) def test_use_get(self): - base, sub, related = (self.tables.base, - self.tables.sub, - self.tables.related) + base, sub, related = ( + self.tables.base, + self.tables.sub, + self.tables.related, + ) # test [ticket:1186] class Base(fixtures.BasicEntity): @@ -903,22 +1002,27 @@ class M2OUseGetTest(fixtures.MappedTest): class Related(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='b') - mapper(Sub, sub, inherits=Base, polymorphic_identity='s') - mapper(Related, related, properties={ - # previously, this was needed for the comparison to occur: - # the 'primaryjoin' looks just like "Sub"'s "get" clause - # (based on the Base id), and foreign_keys since that join - # condition doesn't actually have any fks in it - # 'sub':relationship(Sub, primaryjoin=base.c.id==related.c.sub_id, - # foreign_keys=related.c.sub_id) - # now we can use this: - 'sub': relationship(Sub) - }) + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="b" + ) + mapper(Sub, sub, inherits=Base, polymorphic_identity="s") + mapper( + Related, + related, + properties={ + # previously, this was needed for the comparison to occur: + # the 'primaryjoin' looks just like "Sub"'s "get" clause + # (based on the Base id), and foreign_keys since that join + # condition doesn't actually have any fks in it + # 'sub':relationship(Sub, primaryjoin=base.c.id==related.c.sub_id, + # foreign_keys=related.c.sub_id) + # now we can use this: + "sub": relationship(Sub) + }, + ) - assert class_mapper(Related).get_property('sub').strategy.use_get + assert class_mapper(Related).get_property("sub").strategy.use_get sess = create_session() s1 = Sub() @@ -932,6 +1036,7 @@ class M2OUseGetTest(fixtures.MappedTest): def go(): assert r1.sub + self.assert_sql_count(testing.db, go, 0) @@ -939,23 +1044,36 @@ class GetTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global foo, bar, blub - foo = Table('foo', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(30)), - Column('data', String(20))) - - bar = Table('bar', metadata, - Column('id', Integer, ForeignKey( - 'foo.id'), primary_key=True), - Column('bar_data', String(20))) - - blub = Table('blub', metadata, - Column('blub_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('foo_id', Integer, ForeignKey('foo.id')), - Column('bar_id', Integer, ForeignKey('bar.id')), - Column('blub_data', String(20))) + foo = Table( + "foo", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(30)), + Column("data", String(20)), + ) + + bar = Table( + "bar", + metadata, + Column("id", Integer, ForeignKey("foo.id"), primary_key=True), + Column("bar_data", String(20)), + ) + + blub = Table( + "blub", + metadata, + Column( + "blub_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("foo_id", Integer, ForeignKey("foo.id")), + Column("bar_id", Integer, ForeignKey("bar.id")), + Column("blub_data", String(20)), + ) @classmethod def setup_classes(cls): @@ -975,18 +1093,21 @@ class GetTest(fixtures.MappedTest): self._do_get_test(False) def _do_get_test(self, polymorphic): - foo, Bar, Blub, blub, bar, Foo = (self.tables.foo, - self.classes.Bar, - self.classes.Blub, - self.tables.blub, - self.tables.bar, - self.classes.Foo) + foo, Bar, Blub, blub, bar, Foo = ( + self.tables.foo, + self.classes.Bar, + self.classes.Blub, + self.tables.blub, + self.tables.bar, + self.classes.Foo, + ) if polymorphic: - mapper(Foo, foo, polymorphic_on=foo.c.type, - polymorphic_identity='foo') - mapper(Bar, bar, inherits=Foo, polymorphic_identity='bar') - mapper(Blub, blub, inherits=Bar, polymorphic_identity='blub') + mapper( + Foo, foo, polymorphic_on=foo.c.type, polymorphic_identity="foo" + ) + mapper(Bar, bar, inherits=Foo, polymorphic_identity="bar") + mapper(Blub, blub, inherits=Bar, polymorphic_identity="blub") else: mapper(Foo, foo) mapper(Bar, bar, inherits=Foo) @@ -1002,6 +1123,7 @@ class GetTest(fixtures.MappedTest): sess.flush() if polymorphic: + def go(): assert sess.query(Foo).get(f.id) is f assert sess.query(Foo).get(b.id) is b @@ -1048,18 +1170,27 @@ class EagerLazyTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global foo, bar, bar_foo - foo = Table('foo', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(30))) - bar = Table('bar', metadata, - Column('id', Integer, ForeignKey( - 'foo.id'), primary_key=True), - Column('bar_data', String(30))) - - bar_foo = Table('bar_foo', metadata, - Column('bar_id', Integer, ForeignKey('bar.id')), - Column('foo_id', Integer, ForeignKey('foo.id'))) + foo = Table( + "foo", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(30)), + ) + bar = Table( + "bar", + metadata, + Column("id", Integer, ForeignKey("foo.id"), primary_key=True), + Column("bar_data", String(30)), + ) + + bar_foo = Table( + "bar_foo", + metadata, + Column("bar_id", Integer, ForeignKey("bar.id")), + Column("foo_id", Integer, ForeignKey("foo.id")), + ) def test_basic(self): class Foo(object): @@ -1070,17 +1201,17 @@ class EagerLazyTest(fixtures.MappedTest): foos = mapper(Foo, foo) bars = mapper(Bar, bar, inherits=foos) - bars.add_property('lazy', relationship(foos, bar_foo, lazy='select')) - bars.add_property('eager', relationship(foos, bar_foo, lazy='joined')) + bars.add_property("lazy", relationship(foos, bar_foo, lazy="select")) + bars.add_property("eager", relationship(foos, bar_foo, lazy="joined")) - foo.insert().execute(data='foo1') - bar.insert().execute(id=1, data='bar1') + foo.insert().execute(data="foo1") + bar.insert().execute(id=1, data="bar1") - foo.insert().execute(data='foo2') - bar.insert().execute(id=2, data='bar2') + foo.insert().execute(data="foo2") + bar.insert().execute(id=2, data="bar2") - foo.insert().execute(data='foo3') # 3 - foo.insert().execute(data='foo4') # 4 + foo.insert().execute(data="foo3") # 3 + foo.insert().execute(data="foo4") # 4 bar_foo.insert().execute(bar_id=1, foo_id=3) bar_foo.insert().execute(bar_id=2, foo_id=4) @@ -1097,16 +1228,21 @@ class EagerTargetingTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('a_table', metadata, - Column('id', Integer, primary_key=True), - Column('name', String(50)), - Column('type', String(30), nullable=False), - Column('parent_id', Integer, ForeignKey('a_table.id'))) + Table( + "a_table", + metadata, + Column("id", Integer, primary_key=True), + Column("name", String(50)), + Column("type", String(30), nullable=False), + Column("parent_id", Integer, ForeignKey("a_table.id")), + ) - Table('b_table', metadata, - Column('id', Integer, ForeignKey( - 'a_table.id'), primary_key=True), - Column('b_data', String(50))) + Table( + "b_table", + metadata, + Column("id", Integer, ForeignKey("a_table.id"), primary_key=True), + Column("b_data", String(50)), + ) def test_adapt_stringency(self): b_table, a_table = self.tables.b_table, self.tables.a_table @@ -1118,21 +1254,30 @@ class EagerTargetingTest(fixtures.MappedTest): pass mapper( - A, a_table, polymorphic_on=a_table.c.type, - polymorphic_identity='A', - properties={'children': relationship(A, order_by=a_table.c.name)}) + A, + a_table, + polymorphic_on=a_table.c.type, + polymorphic_identity="A", + properties={"children": relationship(A, order_by=a_table.c.name)}, + ) - mapper(B, b_table, inherits=A, polymorphic_identity='B', properties={ - 'b_derived': column_property(b_table.c.b_data + "DATA") - }) + mapper( + B, + b_table, + inherits=A, + polymorphic_identity="B", + properties={ + "b_derived": column_property(b_table.c.b_data + "DATA") + }, + ) sess = create_session() - b1 = B(id=1, name='b1', b_data='i') + b1 = B(id=1, name="b1", b_data="i") sess.add(b1) sess.flush() - b2 = B(id=2, name='b2', b_data='l', parent_id=1) + b2 = B(id=2, name="b2", b_data="l", parent_id=1) sess.add(b2) sess.flush() @@ -1140,14 +1285,18 @@ class EagerTargetingTest(fixtures.MappedTest): sess.expunge_all() node = sess.query(B).filter(B.id == bid).all()[0] - eq_(node, B(id=1, name='b1', b_data='i')) - eq_(node.children[0], B(id=2, name='b2', b_data='l')) + eq_(node, B(id=1, name="b1", b_data="i")) + eq_(node.children[0], B(id=2, name="b2", b_data="l")) sess.expunge_all() - node = sess.query(B).options(joinedload(B.children))\ - .filter(B.id == bid).all()[0] - eq_(node, B(id=1, name='b1', b_data='i')) - eq_(node.children[0], B(id=2, name='b2', b_data='l')) + node = ( + sess.query(B) + .options(joinedload(B.children)) + .filter(B.id == bid) + .all()[0] + ) + eq_(node, B(id=1, name="b1", b_data="i")) + eq_(node.children[0], B(id=2, name="b2", b_data="l")) class FlushTest(fixtures.MappedTest): @@ -1155,34 +1304,55 @@ class FlushTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('users', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('email', String(128)), - Column('password', String(16))) - - Table('roles', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('description', String(32))) - - Table('user_roles', metadata, - Column('user_id', Integer, ForeignKey( - 'users.id'), primary_key=True), - Column('role_id', Integer, ForeignKey( - 'roles.id'), primary_key=True) - ) - - Table('admins', metadata, - Column('admin_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('user_id', Integer, ForeignKey('users.id'))) + Table( + "users", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("email", String(128)), + Column("password", String(16)), + ) + + Table( + "roles", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("description", String(32)), + ) + + Table( + "user_roles", + metadata, + Column( + "user_id", Integer, ForeignKey("users.id"), primary_key=True + ), + Column( + "role_id", Integer, ForeignKey("roles.id"), primary_key=True + ), + ) + + Table( + "admins", + metadata, + Column( + "admin_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("user_id", Integer, ForeignKey("users.id")), + ) def test_one(self): - admins, users, roles, user_roles = (self.tables.admins, - self.tables.users, - self.tables.roles, - self.tables.user_roles) + admins, users, roles, user_roles = ( + self.tables.admins, + self.tables.users, + self.tables.roles, + self.tables.user_roles, + ) class User(object): pass @@ -1192,9 +1362,17 @@ class FlushTest(fixtures.MappedTest): class Admin(User): pass + role_mapper = mapper(Role, roles) - user_mapper = mapper(User, users, properties={ - 'roles': relationship(Role, secondary=user_roles, lazy='joined')}) + user_mapper = mapper( + User, + users, + properties={ + "roles": relationship( + Role, secondary=user_roles, lazy="joined" + ) + }, + ) admin_mapper = mapper(Admin, admins, inherits=user_mapper) sess = create_session() adminrole = Role() @@ -1207,17 +1385,19 @@ class FlushTest(fixtures.MappedTest): # off and insert the many to many row twice. a = Admin() a.roles.append(adminrole) - a.password = 'admin' + a.password = "admin" sess.add(a) sess.flush() - eq_(select([func.count('*')]).select_from(user_roles).scalar(), 1) + eq_(select([func.count("*")]).select_from(user_roles).scalar(), 1) def test_two(self): - admins, users, roles, user_roles = (self.tables.admins, - self.tables.users, - self.tables.roles, - self.tables.user_roles) + admins, users, roles, user_roles = ( + self.tables.admins, + self.tables.users, + self.tables.roles, + self.tables.user_roles, + ) class User(object): def __init__(self, email=None, password=None): @@ -1232,51 +1412,64 @@ class FlushTest(fixtures.MappedTest): pass role_mapper = mapper(Role, roles) - user_mapper = mapper(User, users, properties={ - 'roles': relationship(Role, secondary=user_roles, lazy='joined')}) + user_mapper = mapper( + User, + users, + properties={ + "roles": relationship( + Role, secondary=user_roles, lazy="joined" + ) + }, + ) admin_mapper = mapper(Admin, admins, inherits=user_mapper) # create roles - adminrole = Role('admin') + adminrole = Role("admin") sess = create_session() sess.add(adminrole) sess.flush() # create admin user - a = Admin(email='tim', password='admin') + a = Admin(email="tim", password="admin") a.roles.append(adminrole) sess.add(a) sess.flush() - a.password = 'sadmin' + a.password = "sadmin" sess.flush() - eq_(select([func.count('*')]).select_from(user_roles).scalar(), 1) + eq_(select([func.count("*")]).select_from(user_roles).scalar(), 1) class PassiveDeletesTest(fixtures.MappedTest): - __requires__ = ('foreign_keys',) + __requires__ = ("foreign_keys",) @classmethod def define_tables(cls, metadata): Table( - "a", metadata, - Column('id', Integer, primary_key=True), - Column('type', String(30)) + "a", + metadata, + Column("id", Integer, primary_key=True), + Column("type", String(30)), ) Table( - "b", metadata, + "b", + metadata, Column( - 'id', Integer, ForeignKey('a.id', ondelete="CASCADE"), - primary_key=True), - Column('data', String(10)) + "id", + Integer, + ForeignKey("a.id", ondelete="CASCADE"), + primary_key=True, + ), + Column("data", String(10)), ) Table( - "c", metadata, - Column('cid', Integer, primary_key=True), - Column('bid', ForeignKey('b.id', ondelete="CASCADE")) + "c", + metadata, + Column("cid", Integer, primary_key=True), + Column("bid", ForeignKey("b.id", ondelete="CASCADE")), ) @classmethod @@ -1295,12 +1488,14 @@ class PassiveDeletesTest(fixtures.MappedTest): a, b, c = self.tables("a", "b", "c") mapper( - A, a, passive_deletes=a_p, - polymorphic_on=a.c.type, polymorphic_identity='a') - mapper( - B, b, inherits=A, passive_deletes=b_p, polymorphic_identity='b') - mapper( - C, c, inherits=B, passive_deletes=c_p, polymorphic_identity='c') + A, + a, + passive_deletes=a_p, + polymorphic_on=a.c.type, + polymorphic_identity="a", + ) + mapper(B, b, inherits=A, passive_deletes=b_p, polymorphic_identity="b") + mapper(C, c, inherits=B, passive_deletes=c_p, polymorphic_identity="c") def test_none(self): A, B, C = self.classes("A", "B", "C") @@ -1319,22 +1514,11 @@ class PassiveDeletesTest(fixtures.MappedTest): s.flush() asserter.assert_( RegexSQL( - "SELECT .* " - "FROM c WHERE :param_1 = c.bid", - [{'param_1': 3}] - ), - CompiledSQL( - "DELETE FROM c WHERE c.cid = :cid", - [{'cid': 1}] - ), - CompiledSQL( - "DELETE FROM b WHERE b.id = :id", - [{'id': 3}] + "SELECT .* " "FROM c WHERE :param_1 = c.bid", [{"param_1": 3}] ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 3}] - ) + CompiledSQL("DELETE FROM c WHERE c.cid = :cid", [{"cid": 1}]), + CompiledSQL("DELETE FROM b WHERE b.id = :id", [{"id": 3}]), + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 3}]), ) def test_c_only(self): @@ -1354,12 +1538,9 @@ class PassiveDeletesTest(fixtures.MappedTest): CompiledSQL( "SELECT a.id AS a_id, a.type AS a_type " "FROM a WHERE a.id = :param_1", - [{'param_1': 1}] + [{"param_1": 1}], ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 1}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 1}]), ) b1.id @@ -1367,14 +1548,8 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM b WHERE b.id = :id", - [{'id': 2}] - ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 2}] - ) + CompiledSQL("DELETE FROM b WHERE b.id = :id", [{"id": 2}]), + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 2}]), ) # want to see if the 'C' table loads even though @@ -1384,14 +1559,8 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM b WHERE b.id = :id", - [{'id': 3}] - ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 3}] - ) + CompiledSQL("DELETE FROM b WHERE b.id = :id", [{"id": 3}]), + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 3}]), ) def test_b_only(self): @@ -1411,12 +1580,9 @@ class PassiveDeletesTest(fixtures.MappedTest): CompiledSQL( "SELECT a.id AS a_id, a.type AS a_type " "FROM a WHERE a.id = :param_1", - [{'param_1': 1}] + [{"param_1": 1}], ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 1}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 1}]), ) b1.id @@ -1424,10 +1590,7 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 2}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 2}]) ) c1.id @@ -1435,10 +1598,7 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 3}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 3}]) ) def test_a_only(self): @@ -1458,12 +1618,9 @@ class PassiveDeletesTest(fixtures.MappedTest): CompiledSQL( "SELECT a.id AS a_id, a.type AS a_type " "FROM a WHERE a.id = :param_1", - [{'param_1': 1}] + [{"param_1": 1}], ), - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 1}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 1}]), ) b1.id @@ -1471,10 +1628,7 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 2}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 2}]) ) # want to see if the 'C' table loads even though @@ -1484,10 +1638,7 @@ class PassiveDeletesTest(fixtures.MappedTest): with self.sql_execution_asserter(testing.db) as asserter: s.flush() asserter.assert_( - CompiledSQL( - "DELETE FROM a WHERE a.id = :id", - [{'id': 3}] - ) + CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 3}]) ) @@ -1497,14 +1648,17 @@ class OptimizedGetOnDeferredTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table( - "a", metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) + "a", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), ) Table( - "b", metadata, - Column('id', Integer, ForeignKey('a.id'), primary_key=True), - Column('data', String(10)) + "b", + metadata, + Column("id", Integer, ForeignKey("a.id"), primary_key=True), + Column("data", String(10)), ) @classmethod @@ -1521,39 +1675,48 @@ class OptimizedGetOnDeferredTest(fixtures.MappedTest): a, b = cls.tables("a", "b") mapper(A, a) - mapper(B, b, inherits=A, properties={ - 'data': deferred(b.c.data), - 'expr': column_property(b.c.data + 'q', deferred=True) - }) + mapper( + B, + b, + inherits=A, + properties={ + "data": deferred(b.c.data), + "expr": column_property(b.c.data + "q", deferred=True), + }, + ) def test_column_property(self): A, B = self.classes("A", "B") sess = Session() - b1 = B(data='x') + b1 = B(data="x") sess.add(b1) sess.flush() - eq_(b1.expr, 'xq') + eq_(b1.expr, "xq") def test_expired_column(self): A, B = self.classes("A", "B") sess = Session() - b1 = B(data='x') + b1 = B(data="x") sess.add(b1) sess.flush() - sess.expire(b1, ['data']) + sess.expire(b1, ["data"]) - eq_(b1.data, 'x') + eq_(b1.data, "x") class JoinedNoFKSortingTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table("a", metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True)) - Table("b", metadata, Column('id', Integer, primary_key=True)) - Table("c", metadata, Column('id', Integer, primary_key=True)) + Table( + "a", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + ) + Table("b", metadata, Column("id", Integer, primary_key=True)) + Table("c", metadata, Column("id", Integer, primary_key=True)) @classmethod def setup_classes(cls): @@ -1570,12 +1733,20 @@ class JoinedNoFKSortingTest(fixtures.MappedTest): def setup_mappers(cls): A, B, C = cls.classes.A, cls.classes.B, cls.classes.C mapper(A, cls.tables.a) - mapper(B, cls.tables.b, inherits=A, - inherit_condition=cls.tables.a.c.id == cls.tables.b.c.id, - inherit_foreign_keys=cls.tables.b.c.id) - mapper(C, cls.tables.c, inherits=A, - inherit_condition=cls.tables.a.c.id == cls.tables.c.c.id, - inherit_foreign_keys=cls.tables.c.c.id) + mapper( + B, + cls.tables.b, + inherits=A, + inherit_condition=cls.tables.a.c.id == cls.tables.b.c.id, + inherit_foreign_keys=cls.tables.b.c.id, + ) + mapper( + C, + cls.tables.c, + inherits=A, + inherit_condition=cls.tables.a.c.id == cls.tables.c.c.id, + inherit_foreign_keys=cls.tables.c.c.id, + ) def test_ordering(self): B, C = self.classes.B, self.classes.C @@ -1590,41 +1761,52 @@ class JoinedNoFKSortingTest(fixtures.MappedTest): CompiledSQL("INSERT INTO a () VALUES ()", {}), AllOf( CompiledSQL( - "INSERT INTO b (id) VALUES (:id)", - [{"id": 1}, {"id": 3}] + "INSERT INTO b (id) VALUES (:id)", [{"id": 1}, {"id": 3}] ), CompiledSQL( - "INSERT INTO c (id) VALUES (:id)", - [{"id": 2}, {"id": 4}] - ) - ) + "INSERT INTO c (id) VALUES (:id)", [{"id": 2}, {"id": 4}] + ), + ), ) class VersioningTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('base', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('version_id', Integer, nullable=False), - Column('value', String(40)), - Column('discriminator', Integer, nullable=False)) - Table('subtable', metadata, - Column('id', None, ForeignKey('base.id'), primary_key=True), - Column('subdata', String(50))) - Table('stuff', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('parent', Integer, ForeignKey('base.id'))) + Table( + "base", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("version_id", Integer, nullable=False), + Column("value", String(40)), + Column("discriminator", Integer, nullable=False), + ) + Table( + "subtable", + metadata, + Column("id", None, ForeignKey("base.id"), primary_key=True), + Column("subdata", String(50)), + ) + Table( + "stuff", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("parent", Integer, ForeignKey("base.id")), + ) @testing.emits_warning(r".*updated rowcount") @testing.requires.sane_rowcount_w_returning @engines.close_open_connections def test_save_update(self): - subtable, base, stuff = (self.tables.subtable, - self.tables.base, - self.tables.stuff) + subtable, base, stuff = ( + self.tables.subtable, + self.tables.base, + self.tables.stuff, + ) class Base(fixtures.BasicEntity): pass @@ -1634,19 +1816,22 @@ class VersioningTest(fixtures.MappedTest): class Stuff(Base): pass + mapper(Stuff, stuff) - mapper(Base, base, - polymorphic_on=base.c.discriminator, - version_id_col=base.c.version_id, - polymorphic_identity=1, properties={ - 'stuff': relationship(Stuff) - }) + mapper( + Base, + base, + polymorphic_on=base.c.discriminator, + version_id_col=base.c.version_id, + polymorphic_identity=1, + properties={"stuff": relationship(Stuff)}, + ) mapper(Sub, subtable, inherits=Base, polymorphic_identity=2) sess = create_session() - b1 = Base(value='b1') - s1 = Sub(value='sub1', subdata='some subdata') + b1 = Base(value="b1") + s1 = Sub(value="sub1", subdata="some subdata") sess.add(b1) sess.add(s1) @@ -1654,15 +1839,17 @@ class VersioningTest(fixtures.MappedTest): sess2 = create_session() s2 = sess2.query(Base).get(s1.id) - s2.subdata = 'sess2 subdata' + s2.subdata = "sess2 subdata" - s1.subdata = 'sess1 subdata' + s1.subdata = "sess1 subdata" sess.flush() - assert_raises(orm_exc.StaleDataError, - sess2.query(Base).with_lockmode('read').get, - s1.id) + assert_raises( + orm_exc.StaleDataError, + sess2.query(Base).with_lockmode("read").get, + s1.id, + ) if not testing.db.dialect.supports_sane_rowcount: sess2.flush() @@ -1671,8 +1858,8 @@ class VersioningTest(fixtures.MappedTest): sess2.refresh(s2) if testing.db.dialect.supports_sane_rowcount: - assert s2.subdata == 'sess1 subdata' - s2.subdata = 'sess2 subdata' + assert s2.subdata == "sess1 subdata" + s2.subdata = "sess2 subdata" sess2.flush() @testing.emits_warning(r".*(update|delete)d rowcount") @@ -1686,16 +1873,20 @@ class VersioningTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, - polymorphic_on=base.c.discriminator, - version_id_col=base.c.version_id, polymorphic_identity=1) + mapper( + Base, + base, + polymorphic_on=base.c.discriminator, + version_id_col=base.c.version_id, + polymorphic_identity=1, + ) mapper(Sub, subtable, inherits=Base, polymorphic_identity=2) sess = create_session() - b1 = Base(value='b1') - s1 = Sub(value='sub1', subdata='some subdata') - s2 = Sub(value='sub2', subdata='some other subdata') + b1 = Base(value="b1") + s1 = Sub(value="sub1", subdata="some subdata") + s2 = Sub(value="sub2", subdata="some other subdata") sess.add(b1) sess.add(s1) sess.add(s2) @@ -1707,15 +1898,12 @@ class VersioningTest(fixtures.MappedTest): sess2.delete(s3) sess2.flush() - s2.subdata = 'some new subdata' + s2.subdata = "some new subdata" sess.flush() - s1.subdata = 'some new subdata' + s1.subdata = "some new subdata" if testing.db.dialect.supports_sane_rowcount: - assert_raises( - orm_exc.StaleDataError, - sess.flush - ) + assert_raises(orm_exc.StaleDataError, sess.flush) else: sess.flush() @@ -1724,24 +1912,31 @@ class DistinctPKTest(fixtures.MappedTest): """test the construction of mapper.primary_key when an inheriting relationship joins on a column other than primary key column.""" - run_inserts = 'once' + run_inserts = "once" run_deletes = None @classmethod def define_tables(cls, metadata): global person_table, employee_table, Person, Employee - person_table = Table("persons", metadata, - Column("id", Integer, primary_key=True, - test_needs_autoincrement=True), - Column("name", String(80))) + person_table = Table( + "persons", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(80)), + ) - employee_table = Table("employees", metadata, - Column("eid", Integer, primary_key=True, - test_needs_autoincrement=True), - Column("salary", Integer), - Column("person_id", Integer, - ForeignKey("persons.id"))) + employee_table = Table( + "employees", + metadata, + Column( + "eid", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("salary", Integer), + Column("person_id", Integer, ForeignKey("persons.id")), + ) class Person(object): def __init__(self, name): @@ -1753,8 +1948,8 @@ class DistinctPKTest(fixtures.MappedTest): @classmethod def insert_data(cls): person_insert = person_table.insert() - person_insert.execute(id=1, name='alice') - person_insert.execute(id=2, name='bob') + person_insert.execute(id=1, name="alice") + person_insert.execute(id=2, name="bob") employee_insert = employee_table.insert() employee_insert.execute(id=2, salary=250, person_id=1) # alice @@ -1767,29 +1962,42 @@ class DistinctPKTest(fixtures.MappedTest): def test_explicit_props(self): person_mapper = mapper(Person, person_table) - mapper(Employee, employee_table, inherits=person_mapper, - properties={'pid': person_table.c.id, - 'eid': employee_table.c.eid}) + mapper( + Employee, + employee_table, + inherits=person_mapper, + properties={"pid": person_table.c.id, "eid": employee_table.c.eid}, + ) self._do_test(False) def test_explicit_composite_pk(self): person_mapper = mapper(Person, person_table) - mapper(Employee, employee_table, - inherits=person_mapper, - properties=dict(id=[employee_table.c.eid, person_table.c.id]), - primary_key=[person_table.c.id, employee_table.c.eid]) + mapper( + Employee, + employee_table, + inherits=person_mapper, + properties=dict(id=[employee_table.c.eid, person_table.c.id]), + primary_key=[person_table.c.id, employee_table.c.eid], + ) assert_raises_message( sa_exc.SAWarning, r"On mapper Mapper\|Employee\|employees, " "primary key column 'persons.id' is being " "combined with distinct primary key column 'employees.eid' " "in attribute 'id'. Use explicit properties to give " - "each column its own mapped attribute name.", self._do_test, True) + "each column its own mapped attribute name.", + self._do_test, + True, + ) def test_explicit_pk(self): person_mapper = mapper(Person, person_table) - mapper(Employee, employee_table, inherits=person_mapper, - primary_key=[person_table.c.id]) + mapper( + Employee, + employee_table, + inherits=person_mapper, + primary_key=[person_table.c.id], + ) self._do_test(False) def _do_test(self, composite): @@ -1805,8 +2013,8 @@ class DistinctPKTest(fixtures.MappedTest): bob = query.get(2) alice2 = query.get(1) - assert alice1.name == alice2.name == 'alice' - assert bob.name == 'bob' + assert alice1.name == alice2.name == "alice" + assert bob.name == "bob" class SyncCompileTest(fixtures.MappedTest): @@ -1816,28 +2024,42 @@ class SyncCompileTest(fixtures.MappedTest): def define_tables(cls, metadata): global _a_table, _b_table, _c_table - _a_table = Table('a', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data1', String(128))) + _a_table = Table( + "a", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data1", String(128)), + ) - _b_table = Table('b', metadata, - Column('a_id', Integer, ForeignKey( - 'a.id'), primary_key=True), - Column('data2', String(128))) + _b_table = Table( + "b", + metadata, + Column("a_id", Integer, ForeignKey("a.id"), primary_key=True), + Column("data2", String(128)), + ) - _c_table = Table('c', metadata, - # Column('a_id', Integer, ForeignKey('b.a_id'), - # primary_key=True), #works - Column('b_a_id', Integer, ForeignKey( - 'b.a_id'), primary_key=True), - Column('data3', String(128))) + _c_table = Table( + "c", + metadata, + # Column('a_id', Integer, ForeignKey('b.a_id'), + # primary_key=True), #works + Column("b_a_id", Integer, ForeignKey("b.a_id"), primary_key=True), + Column("data3", String(128)), + ) def test_joins(self): - for j1 in (None, _b_table.c.a_id == _a_table.c.id, _a_table.c.id == - _b_table.c.a_id): - for j2 in (None, _b_table.c.a_id == _c_table.c.b_a_id, - _c_table.c.b_a_id == _b_table.c.a_id): + for j1 in ( + None, + _b_table.c.a_id == _a_table.c.id, + _a_table.c.id == _b_table.c.a_id, + ): + for j2 in ( + None, + _b_table.c.a_id == _c_table.c.b_a_id, + _c_table.c.b_a_id == _b_table.c.a_id, + ): self._do_test(j1, j2) for t in reversed(_a_table.metadata.sorted_tables): t.delete().execute().close() @@ -1855,22 +2077,18 @@ class SyncCompileTest(fixtures.MappedTest): pass mapper(A, _a_table) - mapper(B, _b_table, inherits=A, - inherit_condition=j1 - ) - mapper(C, _c_table, inherits=B, - inherit_condition=j2 - ) + mapper(B, _b_table, inherits=A, inherit_condition=j1) + mapper(C, _c_table, inherits=B, inherit_condition=j2) session = create_session() - a = A(data1='a1') + a = A(data1="a1") session.add(a) - b = B(data1='b1', data2='b2') + b = B(data1="b1", data2="b2") session.add(b) - c = C(data1='c1', data2='c2', data3='c3') + c = C(data1="c1", data2="c2", data3="c3") session.add(c) session.flush() @@ -1888,21 +2106,37 @@ class OverrideColKeyTest(fixtures.MappedTest): def define_tables(cls, metadata): global base, subtable, subtable_two - base = Table('base', metadata, - Column('base_id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(255)), - Column('sqlite_fixer', String(10))) - - subtable = Table('subtable', metadata, - Column('base_id', Integer, ForeignKey( - 'base.base_id'), primary_key=True), - Column('subdata', String(255))) - subtable_two = Table('subtable_two', metadata, - Column('base_id', Integer, primary_key=True), - Column('fk_base_id', Integer, - ForeignKey('base.base_id')), - Column('subdata', String(255))) + base = Table( + "base", + metadata, + Column( + "base_id", + Integer, + primary_key=True, + test_needs_autoincrement=True, + ), + Column("data", String(255)), + Column("sqlite_fixer", String(10)), + ) + + subtable = Table( + "subtable", + metadata, + Column( + "base_id", + Integer, + ForeignKey("base.base_id"), + primary_key=True, + ), + Column("subdata", String(255)), + ) + subtable_two = Table( + "subtable_two", + metadata, + Column("base_id", Integer, primary_key=True), + Column("fk_base_id", Integer, ForeignKey("base.base_id")), + Column("subdata", String(255)), + ) def test_plain(self): # control case @@ -1918,8 +2152,8 @@ class OverrideColKeyTest(fixtures.MappedTest): # Sub gets a "base_id" property using the "base_id" # column of both tables. eq_( - class_mapper(Sub).get_property('base_id').columns, - [subtable.c.base_id, base.c.base_id] + class_mapper(Sub).get_property("base_id").columns, + [subtable.c.base_id, base.c.base_id], ) def test_override_explicit(self): @@ -1933,18 +2167,21 @@ class OverrideColKeyTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, properties={ - 'id': base.c.base_id - }) - mapper(Sub, subtable, inherits=Base, properties={ - # this is the manual way to do it, is not really - # possible in declarative - 'id': [base.c.base_id, subtable.c.base_id] - }) + mapper(Base, base, properties={"id": base.c.base_id}) + mapper( + Sub, + subtable, + inherits=Base, + properties={ + # this is the manual way to do it, is not really + # possible in declarative + "id": [base.c.base_id, subtable.c.base_id] + }, + ) eq_( - class_mapper(Sub).get_property('id').columns, - [base.c.base_id, subtable.c.base_id] + class_mapper(Sub).get_property("id").columns, + [base.c.base_id, subtable.c.base_id], ) s1 = Sub() @@ -1961,19 +2198,14 @@ class OverrideColKeyTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, properties={ - 'id': base.c.base_id - }) + mapper(Base, base, properties={"id": base.c.base_id}) mapper(Sub, subtable, inherits=Base) - eq_( - class_mapper(Sub).get_property('id').columns, - [base.c.base_id] - ) + eq_(class_mapper(Sub).get_property("id").columns, [base.c.base_id]) eq_( - class_mapper(Sub).get_property('base_id').columns, - [subtable.c.base_id] + class_mapper(Sub).get_property("base_id").columns, + [subtable.c.base_id], ) s1 = Sub() @@ -2003,14 +2235,16 @@ class OverrideColKeyTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, properties={ - 'id': base.c.base_id - }) + mapper(Base, base, properties={"id": base.c.base_id}) def go(): - mapper(Sub, subtable, inherits=Base, properties={ - 'id': subtable.c.base_id - }) + mapper( + Sub, + subtable, + inherits=Base, + properties={"id": subtable.c.base_id}, + ) + # Sub mapper compilation needs to detect that "base.c.base_id" # is renamed in the inherited mapper as "id", even though # it has its own "id" property. It then generates @@ -2028,11 +2262,12 @@ class OverrideColKeyTest(fixtures.MappedTest): def go(): mapper(Sub, subtable_two, inherits=Base) + assert_raises_message( sa_exc.SAWarning, "Implicitly combining column base.base_id with " "column subtable_two.base_id under attribute 'base_id'", - go + go, ) def test_plain_descriptor(self): @@ -2138,24 +2373,37 @@ class OptimizedLoadTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('base', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50)), - Column('type', String(50)), - Column('counter', Integer, server_default="1")) - Table('sub', metadata, - Column('id', Integer, ForeignKey('base.id'), primary_key=True), - Column('sub', String(50)), - Column('subcounter', Integer, server_default="1"), - Column('subcounter2', Integer, server_default="1")) - Table('subsub', metadata, - Column('id', Integer, ForeignKey('sub.id'), primary_key=True), - Column('subsubcounter2', Integer, server_default="1")) - Table('with_comp', metadata, - Column('id', Integer, ForeignKey('base.id'), primary_key=True), - Column('a', String(10)), - Column('b', String(10))) + Table( + "base", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + Column("type", String(50)), + Column("counter", Integer, server_default="1"), + ) + Table( + "sub", + metadata, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + Column("sub", String(50)), + Column("subcounter", Integer, server_default="1"), + Column("subcounter2", Integer, server_default="1"), + ) + Table( + "subsub", + metadata, + Column("id", Integer, ForeignKey("sub.id"), primary_key=True), + Column("subsubcounter2", Integer, server_default="1"), + ) + Table( + "with_comp", + metadata, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + Column("a", String(10)), + Column("b", String(10)), + ) def test_no_optimize_on_map_to_join(self): base, sub = self.tables.base, self.tables.sub @@ -2170,13 +2418,20 @@ class OptimizedLoadTest(fixtures.MappedTest): pass mapper(Base, base) - mapper(JoinBase, base.outerjoin(sub), properties=util.OrderedDict( - [('id', [base.c.id, sub.c.id]), - ('counter', [base.c.counter, sub.c.subcounter])])) + mapper( + JoinBase, + base.outerjoin(sub), + properties=util.OrderedDict( + [ + ("id", [base.c.id, sub.c.id]), + ("counter", [base.c.counter, sub.c.subcounter]), + ] + ), + ) mapper(SubJoinBase, inherits=JoinBase) sess = Session() - sess.add(Base(data='data')) + sess.add(Base(data="data")) sess.commit() sjb = sess.query(SubJoinBase).one() @@ -2186,10 +2441,11 @@ class OptimizedLoadTest(fixtures.MappedTest): # this should not use the optimized load, # which assumes discrete tables def go(): - eq_(sjb.data, 'data') + eq_(sjb.data, "data") self.assert_sql_execution( - testing.db, go, + testing.db, + go, CompiledSQL( "SELECT base.id AS base_id, sub.id AS sub_id, " "base.counter AS base_counter, " @@ -2197,7 +2453,10 @@ class OptimizedLoadTest(fixtures.MappedTest): "base.data AS base_data, base.type AS base_type, " "sub.sub AS sub_sub, sub.subcounter2 AS sub_subcounter2 " "FROM base LEFT OUTER JOIN sub ON base.id = sub.id " - "WHERE base.id = :param_1", {'param_1': sjb_id})) + "WHERE base.id = :param_1", + {"param_1": sjb_id}, + ), + ) def test_optimized_passes(self): """"test that the 'optimized load' routine doesn't crash when @@ -2211,16 +2470,21 @@ class OptimizedLoadTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) # redefine Sub's "id" to favor the "id" col in the subtable. # "id" is also part of the primary join condition - mapper(Sub, sub, inherits=Base, - polymorphic_identity='sub', - properties={'id': [sub.c.id, base.c.id]}) + mapper( + Sub, + sub, + inherits=Base, + polymorphic_identity="sub", + properties={"id": [sub.c.id, base.c.id]}, + ) sess = sessionmaker()() - s1 = Sub(data='s1data', sub='s1sub') + s1 = Sub(data="s1data", sub="s1sub") sess.add(s1) sess.commit() sess.expunge_all() @@ -2231,7 +2495,7 @@ class OptimizedLoadTest(fixtures.MappedTest): # unloaded. the optimized load needs to return "None" so regular # full-row loading proceeds s1 = sess.query(Base).first() - assert s1.sub == 's1sub' + assert s1.sub == "s1sub" def test_column_expression(self): base, sub = self.tables.base, self.tables.sub @@ -2241,18 +2505,26 @@ class OptimizedLoadTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - mapper(Sub, sub, inherits=Base, polymorphic_identity='sub', - properties={ - 'concat': column_property(sub.c.sub + "|" + sub.c.sub)}) + + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + mapper( + Sub, + sub, + inherits=Base, + polymorphic_identity="sub", + properties={ + "concat": column_property(sub.c.sub + "|" + sub.c.sub) + }, + ) sess = sessionmaker()() - s1 = Sub(data='s1data', sub='s1sub') + s1 = Sub(data="s1data", sub="s1sub") sess.add(s1) sess.commit() sess.expunge_all() s1 = sess.query(Base).first() - assert s1.concat == 's1sub|s1sub' + assert s1.concat == "s1sub|s1sub" def test_column_expression_joined(self): base, sub = self.tables.base, self.tables.sub @@ -2262,15 +2534,23 @@ class OptimizedLoadTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - mapper(Sub, sub, inherits=Base, polymorphic_identity='sub', - properties={ - 'concat': column_property(base.c.data + "|" + sub.c.sub)}) + + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + mapper( + Sub, + sub, + inherits=Base, + polymorphic_identity="sub", + properties={ + "concat": column_property(base.c.data + "|" + sub.c.sub) + }, + ) sess = sessionmaker()() - s1 = Sub(data='s1data', sub='s1sub') - s2 = Sub(data='s2data', sub='s2sub') - s3 = Sub(data='s3data', sub='s3sub') + s1 = Sub(data="s1data", sub="s1sub") + s2 = Sub(data="s2data", sub="s2sub") + s3 = Sub(data="s3data", sub="s3sub") sess.add_all([s1, s2, s3]) sess.commit() sess.expunge_all() @@ -2281,10 +2561,10 @@ class OptimizedLoadTest(fixtures.MappedTest): eq_( sess.query(Base).order_by(Base.id).all(), [ - Sub(data='s1data', sub='s1sub', concat='s1data|s1sub'), - Sub(data='s2data', sub='s2sub', concat='s2data|s2sub'), - Sub(data='s3data', sub='s3sub', concat='s3data|s3sub') - ] + Sub(data="s1data", sub="s1sub", concat="s1data|s1sub"), + Sub(data="s2data", sub="s2sub", concat="s2data|s2sub"), + Sub(data="s3data", sub="s3sub", concat="s3data|s3sub"), + ], ) def test_composite_column_joined(self): @@ -2306,22 +2586,28 @@ class OptimizedLoadTest(fixtures.MappedTest): def __eq__(self, other): return (self.a == other.a) and (self.b == other.b) - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - mapper(WithComp, with_comp, inherits=Base, polymorphic_identity='wc', - properties={'comp': composite(Comp, - with_comp.c.a, with_comp.c.b)}) + + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + mapper( + WithComp, + with_comp, + inherits=Base, + polymorphic_identity="wc", + properties={"comp": composite(Comp, with_comp.c.a, with_comp.c.b)}, + ) sess = sessionmaker()() - s1 = WithComp(data='s1data', comp=Comp('ham', 'cheese')) - s2 = WithComp(data='s2data', comp=Comp('bacon', 'eggs')) + s1 = WithComp(data="s1data", comp=Comp("ham", "cheese")) + s2 = WithComp(data="s2data", comp=Comp("bacon", "eggs")) sess.add_all([s1, s2]) sess.commit() sess.expunge_all() s1test, s2test = sess.query(Base).order_by(Base.id).all() assert s1test.comp assert s2test.comp - eq_(s1test.comp, Comp('ham', 'cheese')) - eq_(s2test.comp, Comp('bacon', 'eggs')) + eq_(s1test.comp, Comp("ham", "cheese")) + eq_(s2test.comp, Comp("bacon", "eggs")) def test_load_expired_on_pending(self): base, sub = self.tables.base, self.tables.sub @@ -2331,35 +2617,41 @@ class OptimizedLoadTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - mapper(Sub, sub, inherits=Base, polymorphic_identity='sub') + + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + mapper(Sub, sub, inherits=Base, polymorphic_identity="sub") sess = Session() - s1 = Sub(data='s1') + s1 = Sub(data="s1") sess.add(s1) self.assert_sql_execution( testing.db, sess.flush, CompiledSQL( "INSERT INTO base (data, type) VALUES (:data, :type)", - [{'data': 's1', 'type': 'sub'}] + [{"data": "s1", "type": "sub"}], ), CompiledSQL( "INSERT INTO sub (id, sub) VALUES (:id, :sub)", - lambda ctx: {'id': s1.id, 'sub': None} + lambda ctx: {"id": s1.id, "sub": None}, ), ) def go(): eq_(s1.subcounter2, 1) + self.assert_sql_execution( - testing.db, go, + testing.db, + go, CompiledSQL( "SELECT base.counter AS base_counter, " "sub.subcounter AS sub_subcounter, " "sub.subcounter2 AS sub_subcounter2 FROM base JOIN sub " "ON base.id = sub.id WHERE base.id = :param_1", - lambda ctx: {'param_1': s1.id})) + lambda ctx: {"param_1": s1.id}, + ), + ) def test_dont_generate_on_none(self): base, sub = self.tables.base, self.tables.sub @@ -2369,30 +2661,46 @@ class OptimizedLoadTest(fixtures.MappedTest): class Sub(Base): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - m = mapper(Sub, sub, inherits=Base, polymorphic_identity='sub') + + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + m = mapper(Sub, sub, inherits=Base, polymorphic_identity="sub") s1 = Sub() - assert m._optimized_get_statement(attributes.instance_state(s1), - ['subcounter2']) is None + assert ( + m._optimized_get_statement( + attributes.instance_state(s1), ["subcounter2"] + ) + is None + ) # loads s1.id as None eq_(s1.id, None) # this now will come up with a value of None for id - should reject - assert m._optimized_get_statement(attributes.instance_state(s1), - ['subcounter2']) is None + assert ( + m._optimized_get_statement( + attributes.instance_state(s1), ["subcounter2"] + ) + is None + ) s1.id = 1 attributes.instance_state(s1)._commit_all(s1.__dict__, None) - assert m._optimized_get_statement(attributes.instance_state(s1), - ['subcounter2']) is not None + assert ( + m._optimized_get_statement( + attributes.instance_state(s1), ["subcounter2"] + ) + is not None + ) def test_load_expired_on_pending_twolevel(self): - base, sub, subsub = (self.tables.base, - self.tables.sub, - self.tables.subsub) + base, sub, subsub = ( + self.tables.base, + self.tables.sub, + self.tables.subsub, + ) class Base(fixtures.BasicEntity): pass @@ -2403,12 +2711,13 @@ class OptimizedLoadTest(fixtures.MappedTest): class SubSub(Sub): pass - mapper(Base, base, polymorphic_on=base.c.type, - polymorphic_identity='base') - mapper(Sub, sub, inherits=Base, polymorphic_identity='sub') - mapper(SubSub, subsub, inherits=Sub, polymorphic_identity='subsub') + mapper( + Base, base, polymorphic_on=base.c.type, polymorphic_identity="base" + ) + mapper(Sub, sub, inherits=Base, polymorphic_identity="sub") + mapper(SubSub, subsub, inherits=Sub, polymorphic_identity="subsub") sess = Session() - s1 = SubSub(data='s1', counter=1, subcounter=2) + s1 = SubSub(data="s1", counter=1, subcounter=2) sess.add(s1) self.assert_sql_execution( testing.db, @@ -2416,23 +2725,22 @@ class OptimizedLoadTest(fixtures.MappedTest): CompiledSQL( "INSERT INTO base (data, type, counter) VALUES " "(:data, :type, :counter)", - [{'data': 's1', 'type': 'subsub', 'counter': 1}] + [{"data": "s1", "type": "subsub", "counter": 1}], ), CompiledSQL( "INSERT INTO sub (id, sub, subcounter) VALUES " "(:id, :sub, :subcounter)", - lambda ctx: [{'subcounter': 2, 'sub': None, 'id': s1.id}] + lambda ctx: [{"subcounter": 2, "sub": None, "id": s1.id}], ), CompiledSQL( "INSERT INTO subsub (id) VALUES (:id)", - lambda ctx: {'id': s1.id} + lambda ctx: {"id": s1.id}, ), ) def go(): - eq_( - s1.subcounter2, 1 - ) + eq_(s1.subcounter2, 1) + self.assert_sql_execution( testing.db, go, @@ -2441,27 +2749,28 @@ class OptimizedLoadTest(fixtures.MappedTest): "SELECT subsub.subsubcounter2 AS subsub_subsubcounter2, " "sub.subcounter2 AS sub_subcounter2 FROM subsub, sub " "WHERE :param_1 = sub.id AND sub.id = subsub.id", - lambda ctx: {'param_1': s1.id} + lambda ctx: {"param_1": s1.id}, ), CompiledSQL( "SELECT sub.subcounter2 AS sub_subcounter2, " "subsub.subsubcounter2 AS subsub_subsubcounter2 " "FROM sub, subsub " "WHERE :param_1 = sub.id AND sub.id = subsub.id", - lambda ctx: {'param_1': s1.id} + lambda ctx: {"param_1": s1.id}, ), - ) + ), ) class NoPKOnSubTableWarningTest(fixtures.TestBase): - def _fixture(self): metadata = MetaData() - parent = Table('parent', metadata, - Column('id', Integer, primary_key=True)) - child = Table('child', metadata, - Column('id', Integer, ForeignKey('parent.id'))) + parent = Table( + "parent", metadata, Column("id", Integer, primary_key=True) + ) + child = Table( + "child", metadata, Column("id", Integer, ForeignKey("parent.id")) + ) return parent, child def tearDown(self): @@ -2481,7 +2790,10 @@ class NoPKOnSubTableWarningTest(fixtures.TestBase): sa_exc.SAWarning, "Could not assemble any primary keys for locally mapped " "table 'child' - no rows will be persisted in this Table.", - mapper, C, child, inherits=P + mapper, + C, + child, + inherits=P, ) def test_no_warning_with_explicit(self): @@ -2501,13 +2813,15 @@ class NoPKOnSubTableWarningTest(fixtures.TestBase): class InhCondTest(fixtures.TestBase): def test_inh_cond_nonexistent_table_unrelated(self): metadata = MetaData() - base_table = Table("base", metadata, - Column("id", Integer, primary_key=True)) - derived_table = Table("derived", metadata, - Column("id", Integer, ForeignKey( - "base.id"), primary_key=True), - Column("owner_id", Integer, - ForeignKey("owner.owner_id"))) + base_table = Table( + "base", metadata, Column("id", Integer, primary_key=True) + ) + derived_table = Table( + "derived", + metadata, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + Column("owner_id", Integer, ForeignKey("owner.owner_id")), + ) class Base(object): pass @@ -2517,23 +2831,23 @@ class InhCondTest(fixtures.TestBase): mapper(Base, base_table) # succeeds, despite "owner" table not configured yet - m2 = mapper(Derived, derived_table, - inherits=Base) + m2 = mapper(Derived, derived_table, inherits=Base) assert m2.inherit_condition.compare( base_table.c.id == derived_table.c.id ) def test_inh_cond_nonexistent_col_unrelated(self): m = MetaData() - base_table = Table("base", m, - Column("id", Integer, primary_key=True)) - derived_table = Table("derived", m, - Column("id", Integer, ForeignKey('base.id'), - primary_key=True), - Column('order_id', Integer, - ForeignKey('order.foo'))) - order_table = Table('order', m, Column( - 'id', Integer, primary_key=True)) + base_table = Table("base", m, Column("id", Integer, primary_key=True)) + derived_table = Table( + "derived", + m, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + Column("order_id", Integer, ForeignKey("order.foo")), + ) + order_table = Table( + "order", m, Column("id", Integer, primary_key=True) + ) class Base(object): pass @@ -2551,10 +2865,12 @@ class InhCondTest(fixtures.TestBase): def test_inh_cond_no_fk(self): metadata = MetaData() - base_table = Table("base", metadata, - Column("id", Integer, primary_key=True)) - derived_table = Table("derived", metadata, - Column("id", Integer, primary_key=True)) + base_table = Table( + "base", metadata, Column("id", Integer, primary_key=True) + ) + derived_table = Table( + "derived", metadata, Column("id", Integer, primary_key=True) + ) class Base(object): pass @@ -2568,17 +2884,20 @@ class InhCondTest(fixtures.TestBase): "Can't find any foreign key relationships between " "'base' and 'derived'.", mapper, - Derived, derived_table, inherits=Base + Derived, + derived_table, + inherits=Base, ) def test_inh_cond_nonexistent_table_related(self): m1 = MetaData() m2 = MetaData() - base_table = Table("base", m1, - Column("id", Integer, primary_key=True)) - derived_table = Table("derived", m2, - Column("id", Integer, ForeignKey('base.id'), - primary_key=True)) + base_table = Table("base", m1, Column("id", Integer, primary_key=True)) + derived_table = Table( + "derived", + m2, + Column("id", Integer, ForeignKey("base.id"), primary_key=True), + ) class Base(object): pass @@ -2598,16 +2917,19 @@ class InhCondTest(fixtures.TestBase): "could not find table 'base' with which to generate " "a foreign key to target column 'id'", mapper, - Derived, derived_table, inherits=Base + Derived, + derived_table, + inherits=Base, ) def test_inh_cond_nonexistent_col_related(self): m = MetaData() - base_table = Table("base", m, - Column("id", Integer, primary_key=True)) - derived_table = Table("derived", m, - Column("id", Integer, ForeignKey('base.q'), - primary_key=True)) + base_table = Table("base", m, Column("id", Integer, primary_key=True)) + derived_table = Table( + "derived", + m, + Column("id", Integer, ForeignKey("base.q"), primary_key=True), + ) class Base(object): pass @@ -2623,23 +2945,31 @@ class InhCondTest(fixtures.TestBase): "'base.q' on table " "'derived': table 'base' has no column named 'q'", mapper, - Derived, derived_table, inherits=Base + Derived, + derived_table, + inherits=Base, ) class PKDiscriminatorTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - parents = Table('parents', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('name', String(60))) + parents = Table( + "parents", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("name", String(60)), + ) - children = Table('children', metadata, - Column('id', Integer, ForeignKey('parents.id'), - primary_key=True), - Column('type', Integer, primary_key=True), - Column('name', String(60))) + children = Table( + "children", + metadata, + Column("id", Integer, ForeignKey("parents.id"), primary_key=True), + Column("type", Integer, primary_key=True), + Column("name", String(60)), + ) def test_pk_as_discriminator(self): parents, children = self.tables.parents, self.tables.children @@ -2655,17 +2985,23 @@ class PKDiscriminatorTest(fixtures.MappedTest): class A(Child): pass - mapper(Parent, parents, properties={ - 'children': relationship(Child, backref='parent'), - }) - mapper(Child, children, polymorphic_on=children.c.type, - polymorphic_identity=1) + mapper( + Parent, + parents, + properties={"children": relationship(Child, backref="parent")}, + ) + mapper( + Child, + children, + polymorphic_on=children.c.type, + polymorphic_identity=1, + ) mapper(A, inherits=Child, polymorphic_identity=2) s = create_session() - p = Parent('p1') - a = A('a1') + p = Parent("p1") + a = A("a1") p.children.append(a) s.add(p) s.flush() @@ -2673,22 +3009,26 @@ class PKDiscriminatorTest(fixtures.MappedTest): assert a.id assert a.type == 2 - p.name = 'p1new' - a.name = 'a1new' + p.name = "p1new" + a.name = "a1new" s.flush() s.expire_all() - assert a.name == 'a1new' - assert p.name == 'p1new' + assert a.name == "a1new" + assert p.name == "p1new" class NoPolyIdentInMiddleTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('base', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(50), nullable=False)) + Table( + "base", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(50), nullable=False), + ) @classmethod def setup_classes(cls): @@ -2709,18 +3049,20 @@ class NoPolyIdentInMiddleTest(fixtures.MappedTest): @classmethod def setup_mappers(cls): - A, C, B, E, D, base = (cls.classes.A, - cls.classes.C, - cls.classes.B, - cls.classes.E, - cls.classes.D, - cls.tables.base) + A, C, B, E, D, base = ( + cls.classes.A, + cls.classes.C, + cls.classes.B, + cls.classes.E, + cls.classes.D, + cls.tables.base, + ) mapper(A, base, polymorphic_on=base.c.type) - mapper(B, inherits=A, ) - mapper(C, inherits=B, polymorphic_identity='c') - mapper(D, inherits=B, polymorphic_identity='d') - mapper(E, inherits=A, polymorphic_identity='e') + mapper(B, inherits=A) + mapper(C, inherits=B, polymorphic_identity="c") + mapper(D, inherits=B, polymorphic_identity="d") + mapper(E, inherits=A, polymorphic_identity="e") def test_load_from_middle(self): C, B = self.classes.C, self.classes.B @@ -2728,7 +3070,7 @@ class NoPolyIdentInMiddleTest(fixtures.MappedTest): s = Session() s.add(C()) o = s.query(B).first() - eq_(o.type, 'c') + eq_(o.type, "c") assert isinstance(o, C) def test_load_from_base(self): @@ -2737,30 +3079,27 @@ class NoPolyIdentInMiddleTest(fixtures.MappedTest): s = Session() s.add(C()) o = s.query(A).first() - eq_(o.type, 'c') + eq_(o.type, "c") assert isinstance(o, C) def test_discriminator(self): - C, B, base = (self.classes.C, - self.classes.B, - self.tables.base) + C, B, base = (self.classes.C, self.classes.B, self.tables.base) assert class_mapper(B).polymorphic_on is base.c.type assert class_mapper(C).polymorphic_on is base.c.type def test_load_multiple_from_middle(self): - C, B, E, D, base = (self.classes.C, - self.classes.B, - self.classes.E, - self.classes.D, - self.tables.base) + C, B, E, D, base = ( + self.classes.C, + self.classes.B, + self.classes.E, + self.classes.D, + self.tables.base, + ) s = Session() s.add_all([C(), D(), E()]) - eq_( - s.query(B).order_by(base.c.type).all(), - [C(), D()] - ) + eq_(s.query(B).order_by(base.c.type).all(), [C(), D()]) class DeleteOrphanTest(fixtures.MappedTest): @@ -2775,19 +3114,27 @@ class DeleteOrphanTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): global single, parent - single = Table('single', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(50), nullable=False), - Column('data', String(50)), - Column('parent_id', Integer, ForeignKey( - 'parent.id'), nullable=False), - ) - - parent = Table('parent', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50))) + single = Table( + "single", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(50), nullable=False), + Column("data", String(50)), + Column( + "parent_id", Integer, ForeignKey("parent.id"), nullable=False + ), + ) + + parent = Table( + "parent", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + ) def test_orphan_message(self): class Base(fixtures.BasicEntity): @@ -2799,61 +3146,77 @@ class DeleteOrphanTest(fixtures.MappedTest): class Parent(fixtures.BasicEntity): pass - mapper(Base, single, polymorphic_on=single.c.type, - polymorphic_identity='base') - mapper(SubClass, inherits=Base, polymorphic_identity='sub') - mapper(Parent, parent, properties={ - 'related': relationship(Base, cascade="all, delete-orphan") - }) + mapper( + Base, + single, + polymorphic_on=single.c.type, + polymorphic_identity="base", + ) + mapper(SubClass, inherits=Base, polymorphic_identity="sub") + mapper( + Parent, + parent, + properties={ + "related": relationship(Base, cascade="all, delete-orphan") + }, + ) sess = create_session() - s1 = SubClass(data='s1') + s1 = SubClass(data="s1") sess.add(s1) assert_raises(sa_exc.DBAPIError, sess.flush) class PolymorphicUnionTest(fixtures.TestBase, testing.AssertsCompiledSQL): - __dialect__ = 'default' + __dialect__ = "default" def _fixture(self): - t1 = table('t1', column('c1', Integer), - column('c2', Integer), - column('c3', Integer)) - t2 = table('t2', column('c1', Integer), column('c2', Integer), - column('c3', Integer), - column('c4', Integer)) - t3 = table('t3', column('c1', Integer), - column('c3', Integer), - column('c5', Integer)) + t1 = table( + "t1", + column("c1", Integer), + column("c2", Integer), + column("c3", Integer), + ) + t2 = table( + "t2", + column("c1", Integer), + column("c2", Integer), + column("c3", Integer), + column("c4", Integer), + ) + t3 = table( + "t3", + column("c1", Integer), + column("c3", Integer), + column("c5", Integer), + ) return t1, t2, t3 def test_type_col_present(self): t1, t2, t3 = self._fixture() self.assert_compile( polymorphic_union( - util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]), - 'q1' + util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]), "q1" ), "SELECT t1.c1, t1.c2, t1.c3, CAST(NULL AS INTEGER) AS c4, " "CAST(NULL AS INTEGER) AS c5, 'a' AS q1 FROM t1 UNION ALL " "SELECT t2.c1, t2.c2, t2.c3, t2.c4, CAST(NULL AS INTEGER) AS c5, " "'b' AS q1 FROM t2 UNION ALL SELECT t3.c1, " "CAST(NULL AS INTEGER) AS c2, t3.c3, CAST(NULL AS INTEGER) AS c4, " - "t3.c5, 'c' AS q1 FROM t3" + "t3.c5, 'c' AS q1 FROM t3", ) def test_type_col_non_present(self): t1, t2, t3 = self._fixture() self.assert_compile( polymorphic_union( - util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]), - None + util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]), None ), "SELECT t1.c1, t1.c2, t1.c3, CAST(NULL AS INTEGER) AS c4, " "CAST(NULL AS INTEGER) AS c5 FROM t1 UNION ALL SELECT t2.c1, " "t2.c2, t2.c3, t2.c4, CAST(NULL AS INTEGER) AS c5 FROM t2 " "UNION ALL SELECT t3.c1, CAST(NULL AS INTEGER) AS c2, t3.c3, " - "CAST(NULL AS INTEGER) AS c4, t3.c5 FROM t3" + "CAST(NULL AS INTEGER) AS c4, t3.c5 FROM t3", ) def test_no_cast_null(self): @@ -2861,26 +3224,33 @@ class PolymorphicUnionTest(fixtures.TestBase, testing.AssertsCompiledSQL): self.assert_compile( polymorphic_union( util.OrderedDict([("a", t1), ("b", t2), ("c", t3)]), - 'q1', cast_nulls=False + "q1", + cast_nulls=False, ), "SELECT t1.c1, t1.c2, t1.c3, NULL AS c4, NULL AS c5, 'a' AS q1 " "FROM t1 UNION ALL SELECT t2.c1, t2.c2, t2.c3, t2.c4, NULL AS c5, " "'b' AS q1 FROM t2 UNION ALL SELECT t3.c1, NULL AS c2, t3.c3, " - "NULL AS c4, t3.c5, 'c' AS q1 FROM t3" + "NULL AS c4, t3.c5, 'c' AS q1 FROM t3", ) class NameConflictTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - content = Table('content', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('type', String(30))) - foo = Table('foo', metadata, - Column('id', Integer, ForeignKey('content.id'), - primary_key=True), - Column('content_type', String(30))) + content = Table( + "content", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("type", String(30)), + ) + foo = Table( + "foo", + metadata, + Column("id", Integer, ForeignKey("content.id"), primary_key=True), + Column("content_type", String(30)), + ) def test_name_conflict(self): class Content(object): @@ -2888,15 +3258,20 @@ class NameConflictTest(fixtures.MappedTest): class Foo(Content): pass - mapper(Content, self.tables.content, - polymorphic_on=self.tables.content.c.type) - mapper(Foo, self.tables.foo, inherits=Content, - polymorphic_identity='foo') + + mapper( + Content, + self.tables.content, + polymorphic_on=self.tables.content.c.type, + ) + mapper( + Foo, self.tables.foo, inherits=Content, polymorphic_identity="foo" + ) sess = create_session() f = Foo() - f.content_type = 'bar' + f.content_type = "bar" sess.add(f) sess.flush() f_id = f.id sess.expunge_all() - assert sess.query(Content).get(f_id).content_type == 'bar' + assert sess.query(Content).get(f_id).content_type == "bar" |