diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2019-01-06 01:14:26 -0500 |
|---|---|---|
| committer | mike bayer <mike_mp@zzzcomputing.com> | 2019-01-06 17:34:50 +0000 |
| commit | 1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch) | |
| tree | 28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/orm/inheritance/test_basic.py | |
| parent | 404e69426b05a82d905cbb3ad33adafccddb00dd (diff) | |
| download | sqlalchemy-1e1a38e7801f410f244e4bbb44ec795ae152e04e.tar.gz | |
Run black -l 79 against all source files
This is a straight reformat run using black as is, with no edits
applied at all.
The black run will format code consistently, however in
some cases that are prevalent in SQLAlchemy code it produces
too-long lines. The too-long lines will be resolved in the
following commit that will resolve all remaining flake8 issues
including shadowed builtins, long lines, import order, unused
imports, duplicate imports, and docstring issues.
Change-Id: I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9
Diffstat (limited to 'test/orm/inheritance/test_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" |