diff options
Diffstat (limited to 'test/ext/declarative/test_basic.py')
| -rw-r--r-- | test/ext/declarative/test_basic.py | 1673 |
1 files changed, 934 insertions, 739 deletions
diff --git a/test/ext/declarative/test_basic.py b/test/ext/declarative/test_basic.py index f45421ad9..9de7f5ab9 100644 --- a/test/ext/declarative/test_basic.py +++ b/test/ext/declarative/test_basic.py @@ -1,18 +1,41 @@ - -from sqlalchemy.testing import eq_, assert_raises, \ - assert_raises_message, expect_warnings, is_ +from sqlalchemy.testing import ( + eq_, + assert_raises, + assert_raises_message, + expect_warnings, + is_, +) from sqlalchemy.testing import assertions from sqlalchemy.ext import declarative as decl from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy import exc import sqlalchemy as sa from sqlalchemy import testing, util -from sqlalchemy import MetaData, Integer, String, ForeignKey, \ - ForeignKeyConstraint, Index, UniqueConstraint, CheckConstraint +from sqlalchemy import ( + MetaData, + Integer, + String, + ForeignKey, + ForeignKeyConstraint, + Index, + UniqueConstraint, + CheckConstraint, +) from sqlalchemy.testing.schema import Table, Column -from sqlalchemy.orm import relationship, create_session, class_mapper, \ - joinedload, configure_mappers, backref, clear_mappers, \ - column_property, composite, Session, properties, deferred +from sqlalchemy.orm import ( + relationship, + create_session, + class_mapper, + joinedload, + configure_mappers, + backref, + clear_mappers, + column_property, + composite, + Session, + properties, + deferred, +) from sqlalchemy.util import with_metaclass from sqlalchemy.ext.declarative import declared_attr, synonym_for from sqlalchemy.testing import fixtures, mock @@ -26,10 +49,12 @@ Base = None User = Address = None -class DeclarativeTestBase(fixtures.TestBase, - testing.AssertsExecutionResults, - testing.AssertsCompiledSQL): - __dialect__ = 'default' +class DeclarativeTestBase( + fixtures.TestBase, + testing.AssertsExecutionResults, + testing.AssertsCompiledSQL, +): + __dialect__ = "default" def setup(self): global Base @@ -42,119 +67,132 @@ class DeclarativeTestBase(fixtures.TestBase, class DeclarativeTest(DeclarativeTestBase): - def test_basic(self): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) addresses = relationship("Address", backref="user") class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column(String(50), key='_email') - user_id = Column('user_id', Integer, ForeignKey('users.id'), - key='_user_id') + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column(String(50), key="_email") + user_id = Column( + "user_id", Integer, ForeignKey("users.id"), key="_user_id" + ) Base.metadata.create_all() - eq_(Address.__table__.c['id'].name, 'id') - eq_(Address.__table__.c['_email'].name, 'email') - eq_(Address.__table__.c['_user_id'].name, 'user_id') + eq_(Address.__table__.c["id"].name, "id") + eq_(Address.__table__.c["_email"].name, "email") + eq_(Address.__table__.c["_user_id"].name, "user_id") - u1 = User(name='u1', addresses=[ - Address(email='one'), - Address(email='two'), - ]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [User(name='u1', addresses=[ - Address(email='one'), - Address(email='two'), - ])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) - a1 = sess.query(Address).filter(Address.email == 'two').one() - eq_(a1, Address(email='two')) - eq_(a1.user, User(name='u1')) + a1 = sess.query(Address).filter(Address.email == "two").one() + eq_(a1, Address(email="two")) + eq_(a1.user, User(name="u1")) def test_unicode_string_resolve(self): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) addresses = relationship(util.u("Address"), backref="user") class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column(String(50), key='_email') - user_id = Column('user_id', Integer, ForeignKey('users.id'), - key='_user_id') + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column(String(50), key="_email") + user_id = Column( + "user_id", Integer, ForeignKey("users.id"), key="_user_id" + ) assert User.addresses.property.mapper.class_ is Address def test_unicode_string_resolve_backref(self): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column(String(50), key='_email') - user_id = Column('user_id', Integer, ForeignKey('users.id'), - key='_user_id') + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column(String(50), key="_email") + user_id = Column( + "user_id", Integer, ForeignKey("users.id"), key="_user_id" + ) user = relationship( - User, - backref=backref("addresses", - order_by=util.u("Address.email"))) + User, + backref=backref("addresses", order_by=util.u("Address.email")), + ) assert Address.user.property.mapper.class_ is User def test_no_table(self): def go(): class User(Base): - id = Column('id', Integer, primary_key=True) + id = Column("id", Integer, primary_key=True) - assert_raises_message(sa.exc.InvalidRequestError, - 'does not have a __table__', go) + assert_raises_message( + sa.exc.InvalidRequestError, "does not have a __table__", go + ) def test_table_args_empty_dict(self): - class MyModel(Base): - __tablename__ = 'test' + __tablename__ = "test" id = Column(Integer, primary_key=True) __table_args__ = {} def test_table_args_empty_tuple(self): - class MyModel(Base): - __tablename__ = 'test' + __tablename__ = "test" id = Column(Integer, primary_key=True) __table_args__ = () def test_cant_add_columns(self): t = Table( - 't', Base.metadata, - Column('id', Integer, primary_key=True), - Column('data', String)) + "t", + Base.metadata, + Column("id", Integer, primary_key=True), + Column("data", String), + ) def go(): class User(Base): @@ -163,9 +201,11 @@ class DeclarativeTest(DeclarativeTestBase): # can't specify new columns not already in the table - assert_raises_message(sa.exc.ArgumentError, - "Can't add additional column 'foo' when " - "specifying __table__", go) + assert_raises_message( + sa.exc.ArgumentError, + "Can't add additional column 'foo' when " "specifying __table__", + go, + ) # regular re-mapping works tho @@ -173,16 +213,18 @@ class DeclarativeTest(DeclarativeTestBase): __table__ = t some_data = t.c.data - assert class_mapper(Bar).get_property('some_data').columns[0] \ - is t.c.data + assert ( + class_mapper(Bar).get_property("some_data").columns[0] is t.c.data + ) def test_lower_case_c_column_warning(self): with assertions.expect_warnings( r"Attribute 'x' on class <class .*Foo.* appears to be a " - r"non-schema 'sqlalchemy.sql.column\(\)' object; ", + r"non-schema 'sqlalchemy.sql.column\(\)' object; " ): + class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) x = sa.sql.expression.column(Integer) @@ -194,19 +236,21 @@ class DeclarativeTest(DeclarativeTestBase): with assertions.expect_warnings( r"Attribute 'x' on class <class .*MyMixin.* appears to be a " - r"non-schema 'sqlalchemy.sql.column\(\)' object; ", + r"non-schema 'sqlalchemy.sql.column\(\)' object; " ): + class Foo2(MyMixin, Base): - __tablename__ = 'foo2' + __tablename__ = "foo2" id = Column(Integer, primary_key=True) with assertions.expect_warnings( r"Attribute 'x' on class <class .*Foo3.* appears to be a " - r"non-schema 'sqlalchemy.sql.column\(\)' object; ", + r"non-schema 'sqlalchemy.sql.column\(\)' object; " ): + class Foo3(Base): - __tablename__ = 'foo3' + __tablename__ = "foo3" id = Column(Integer, primary_key=True) @@ -218,8 +262,9 @@ class DeclarativeTest(DeclarativeTestBase): with assertions.expect_warnings( r"Attribute 'x' on class <class .*Foo4.* appears to be a " - r"non-schema 'sqlalchemy.sql.column\(\)' object; ", + r"non-schema 'sqlalchemy.sql.column\(\)' object; " ): + class MyMixin2(object): @declared_attr def x(cls): @@ -228,76 +273,79 @@ class DeclarativeTest(DeclarativeTestBase): y = Column(Integer) class Foo4(MyMixin2, Base): - __tablename__ = 'foo4' + __tablename__ = "foo4" id = Column(Integer, primary_key=True) def test_column_named_twice(self): def go(): class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) - x = Column('x', Integer) - y = Column('x', Integer) + x = Column("x", Integer) + y = Column("x", Integer) + assert_raises_message( sa.exc.SAWarning, "On class 'Foo', Column object 'x' named directly multiple times, " "only one will be used: x, y", - go + go, ) def test_column_repeated_under_prop(self): def go(): class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) - x = Column('x', Integer) + x = Column("x", Integer) y = column_property(x) - z = Column('x', Integer) + z = Column("x", Integer) assert_raises_message( sa.exc.SAWarning, "On class 'Foo', Column object 'x' named directly multiple times, " "only one will be used: x, y, z", - go + go, ) def test_using_explicit_prop_in_schema_objects(self): class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) cprop = column_property(Column(Integer)) - __table_args__ = ( - UniqueConstraint(cprop), - ) + __table_args__ = (UniqueConstraint(cprop),) + uq = [ - c for c in Foo.__table__.constraints - if isinstance(c, UniqueConstraint)][0] + c + for c in Foo.__table__.constraints + if isinstance(c, UniqueConstraint) + ][0] is_(uq.columns.cprop, Foo.__table__.c.cprop) class Bar(Base): - __tablename__ = 'bar' + __tablename__ = "bar" id = Column(Integer, primary_key=True) cprop = deferred(Column(Integer)) - __table_args__ = ( - CheckConstraint(cprop > sa.func.foo()), - ) + __table_args__ = (CheckConstraint(cprop > sa.func.foo()),) + ck = [ - c for c in Bar.__table__.constraints - if isinstance(c, CheckConstraint)][0] + c + for c in Bar.__table__.constraints + if isinstance(c, CheckConstraint) + ][0] is_(ck.columns.cprop, Bar.__table__.c.cprop) if testing.requires.python3.enabled: # test the existing failure case in case something changes def go(): class Bat(Base): - __tablename__ = 'bat' + __tablename__ = "bat" id = Column(Integer, primary_key=True) cprop = deferred(Column(Integer)) @@ -306,43 +354,44 @@ class DeclarativeTest(DeclarativeTestBase): # "cprop > 5" because the column property isn't # a full blown column - __table_args__ = ( - CheckConstraint(cprop > 5), - ) + __table_args__ = (CheckConstraint(cprop > 5),) + assert_raises(TypeError, go) def test_relationship_level_msg_for_invalid_callable(self): class A(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) class B(Base): - __tablename__ = 'b' + __tablename__ = "b" id = Column(Integer, primary_key=True) - a_id = Column(Integer, ForeignKey('a.id')) - a = relationship('a') + a_id = Column(Integer, ForeignKey("a.id")) + a = relationship("a") + assert_raises_message( sa.exc.ArgumentError, "relationship 'a' expects a class or a mapper " "argument .received: .*Table", - configure_mappers + configure_mappers, ) def test_relationship_level_msg_for_invalid_object(self): class A(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) class B(Base): - __tablename__ = 'b' + __tablename__ = "b" id = Column(Integer, primary_key=True) - a_id = Column(Integer, ForeignKey('a.id')) + a_id = Column(Integer, ForeignKey("a.id")) a = relationship(A.__table__) + assert_raises_message( sa.exc.ArgumentError, "relationship 'a' expects a class or a mapper " "argument .received: .*Table", - configure_mappers + configure_mappers, ) def test_difficult_class(self): @@ -350,10 +399,9 @@ class DeclarativeTest(DeclarativeTestBase): # metaclass to mock the way zope.interface breaks getattr() class BrokenMeta(type): - def __getattribute__(self, attr): - if attr == 'xyzzy': - raise AttributeError('xyzzy') + if attr == "xyzzy": + raise AttributeError("xyzzy") else: return object.__getattribute__(self, attr) @@ -364,23 +412,24 @@ class DeclarativeTest(DeclarativeTestBase): # _as_declarative() inspects obj.__class__.__bases__ class User(BrokenParent, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) decl.instrument_declarative(User, {}, Base.metadata) def test_reserved_identifiers(self): def go1(): class User1(Base): - __tablename__ = 'user1' + __tablename__ = "user1" id = Column(Integer, primary_key=True) metadata = Column(Integer) def go2(): class User2(Base): - __tablename__ = 'user2' + __tablename__ = "user2" id = Column(Integer, primary_key=True) metadata = relationship("Address") @@ -390,20 +439,20 @@ class DeclarativeTest(DeclarativeTestBase): "Attribute name 'metadata' is reserved " "for the MetaData instance when using a " "declarative base class.", - go + go, ) def test_undefer_column_name(self): # TODO: not sure if there was an explicit # test for this elsewhere foo = Column(Integer) - eq_(str(foo), '(no name)') + eq_(str(foo), "(no name)") eq_(foo.key, None) eq_(foo.name, None) - decl.base._undefer_column_name('foo', foo) - eq_(str(foo), 'foo') - eq_(foo.key, 'foo') - eq_(foo.name, 'foo') + decl.base._undefer_column_name("foo", foo) + eq_(str(foo), "foo") + eq_(foo.key, "foo") + eq_(foo.name, "foo") def test_recompile_on_othermapper(self): """declarative version of the same test in mappers.py""" @@ -411,19 +460,20 @@ class DeclarativeTest(DeclarativeTestBase): from sqlalchemy.orm import mapperlib class User(Base): - __tablename__ = 'users' + __tablename__ = "users" - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) class Address(Base): - __tablename__ = 'addresses' + __tablename__ = "addresses" - id = Column('id', Integer, primary_key=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) - user = relationship("User", primaryjoin=user_id == User.id, - backref="addresses") + id = Column("id", Integer, primary_key=True) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) + user = relationship( + "User", primaryjoin=user_id == User.id, backref="addresses" + ) assert mapperlib.Mapper._new_mappers is True u = User() # noqa @@ -433,81 +483,99 @@ class DeclarativeTest(DeclarativeTestBase): def test_string_dependency_resolution(self): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) addresses = relationship( - 'Address', - order_by='desc(Address.email)', - primaryjoin='User.id==Address.user_id', - foreign_keys='[Address.user_id]', - backref=backref('user', - primaryjoin='User.id==Address.user_id', - foreign_keys='[Address.user_id]')) + "Address", + order_by="desc(Address.email)", + primaryjoin="User.id==Address.user_id", + foreign_keys="[Address.user_id]", + backref=backref( + "user", + primaryjoin="User.id==Address.user_id", + foreign_keys="[Address.user_id]", + ), + ) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) email = Column(String(50)) user_id = Column(Integer) # note no foreign key Base.metadata.create_all() sess = create_session() u1 = User( - name='ed', addresses=[ - Address(email='abc'), - Address(email='def'), Address(email='xyz')]) + name="ed", + addresses=[ + Address(email="abc"), + Address(email="def"), + Address(email="xyz"), + ], + ) sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).filter(User.name == 'ed').one(), - User(name='ed', addresses=[ - Address(email='xyz'), - Address(email='def'), Address(email='abc')])) + eq_( + sess.query(User).filter(User.name == "ed").one(), + User( + name="ed", + addresses=[ + Address(email="xyz"), + Address(email="def"), + Address(email="abc"), + ], + ), + ) class Foo(Base, fixtures.ComparableEntity): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) - rel = relationship('User', - primaryjoin='User.addresses==Foo.id') + rel = relationship("User", primaryjoin="User.addresses==Foo.id") - assert_raises_message(exc.InvalidRequestError, - "'addresses' is not an instance of " - "ColumnProperty", configure_mappers) + assert_raises_message( + exc.InvalidRequestError, + "'addresses' is not an instance of " "ColumnProperty", + configure_mappers, + ) def test_string_dependency_resolution_synonym(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) Base.metadata.create_all() sess = create_session() - u1 = User(name='ed') + u1 = User(name="ed") sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).filter(User.name == 'ed').one(), - User(name='ed')) + eq_(sess.query(User).filter(User.name == "ed").one(), User(name="ed")) class Foo(Base, fixtures.ComparableEntity): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) _user_id = Column(Integer) - rel = relationship('User', - uselist=False, - foreign_keys=[User.id], - primaryjoin='Foo.user_id==User.id') + rel = relationship( + "User", + uselist=False, + foreign_keys=[User.id], + primaryjoin="Foo.user_id==User.id", + ) - @synonym_for('_user_id') + @synonym_for("_user_id") @property def user_id(self): return self._user_id @@ -520,18 +588,18 @@ class DeclarativeTest(DeclarativeTestBase): from sqlalchemy.ext.hybrid import hybrid_property class User(Base): - __tablename__ = 'user' + __tablename__ = "user" id = Column(Integer, primary_key=True) firstname = Column(String(50)) lastname = Column(String(50)) - game_id = Column(Integer, ForeignKey('game.id')) + game_id = Column(Integer, ForeignKey("game.id")) @hybrid_property def fullname(self): return self.firstname + " " + self.lastname class Game(Base): - __tablename__ = 'game' + __tablename__ = "game" id = Column(Integer, primary_key=True) name = Column(String(50)) users = relationship("User", order_by="User.fullname") @@ -543,17 +611,17 @@ class DeclarativeTest(DeclarativeTestBase): "user_1.id AS user_1_id, user_1.firstname AS user_1_firstname, " "user_1.lastname AS user_1_lastname, " "user_1.game_id AS user_1_game_id " - "FROM game LEFT OUTER JOIN \"user\" AS user_1 ON game.id = " + 'FROM game LEFT OUTER JOIN "user" AS user_1 ON game.id = ' "user_1.game_id ORDER BY " - "user_1.firstname || :firstname_1 || user_1.lastname" + "user_1.firstname || :firstname_1 || user_1.lastname", ) def test_string_dependency_resolution_asselectable(self): class A(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) - b_id = Column(ForeignKey('b.id')) + b_id = Column(ForeignKey("b.id")) d = relationship( "D", @@ -564,22 +632,23 @@ class DeclarativeTest(DeclarativeTestBase): ) class B(Base): - __tablename__ = 'b' + __tablename__ = "b" id = Column(Integer, primary_key=True) - d_id = Column(ForeignKey('d.id')) + d_id = Column(ForeignKey("d.id")) class C(Base): - __tablename__ = 'c' + __tablename__ = "c" id = Column(Integer, primary_key=True) - a_id = Column(ForeignKey('a.id')) - d_id = Column(ForeignKey('d.id')) + a_id = Column(ForeignKey("a.id")) + d_id = Column(ForeignKey("d.id")) class D(Base): - __tablename__ = 'd' + __tablename__ = "d" id = Column(Integer, primary_key=True) + s = Session() self.assert_compile( s.query(A).join(A.d), @@ -590,46 +659,47 @@ class DeclarativeTest(DeclarativeTestBase): ) def test_string_dependency_resolution_no_table(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) class Bar(Base, fixtures.ComparableEntity): - __tablename__ = 'bar' + __tablename__ = "bar" id = Column(Integer, primary_key=True) - rel = relationship('User', - primaryjoin='User.id==Bar.__table__.id') + rel = relationship("User", primaryjoin="User.id==Bar.__table__.id") - assert_raises_message(exc.InvalidRequestError, - "does not have a mapped column named " - "'__table__'", configure_mappers) + assert_raises_message( + exc.InvalidRequestError, + "does not have a mapped column named " "'__table__'", + configure_mappers, + ) def test_string_w_pj_annotations(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) email = Column(String(50)) user_id = Column(Integer) user = relationship( - "User", - primaryjoin="remote(User.id)==foreign(Address.user_id)" + "User", primaryjoin="remote(User.id)==foreign(Address.user_id)" ) eq_( Address.user.property._join_condition.local_remote_pairs, - [(Address.__table__.c.user_id, User.__table__.c.id)] + [(Address.__table__.c.user_id, User.__table__.c.id)], ) def test_string_dependency_resolution_no_magic(self): @@ -637,148 +707,166 @@ class DeclarativeTest(DeclarativeTestBase): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" id = Column(Integer, primary_key=True) addresses = relationship( - 'Address', - primaryjoin='User.id==Address.user_id.prop.columns[0]') + "Address", + primaryjoin="User.id==Address.user_id.prop.columns[0]", + ) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" id = Column(Integer, primary_key=True) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) configure_mappers() - eq_(str(User.addresses.prop.primaryjoin), - 'users.id = addresses.user_id') + eq_( + str(User.addresses.prop.primaryjoin), + "users.id = addresses.user_id", + ) def test_string_dependency_resolution_module_qualified(self): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" id = Column(Integer, primary_key=True) addresses = relationship( - '%s.Address' % __name__, - primaryjoin='%s.User.id==%s.Address.user_id.prop.columns[0]' - % (__name__, __name__)) + "%s.Address" % __name__, + primaryjoin="%s.User.id==%s.Address.user_id.prop.columns[0]" + % (__name__, __name__), + ) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" id = Column(Integer, primary_key=True) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) configure_mappers() - eq_(str(User.addresses.prop.primaryjoin), - 'users.id = addresses.user_id') + eq_( + str(User.addresses.prop.primaryjoin), + "users.id = addresses.user_id", + ) def test_string_dependency_resolution_in_backref(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" id = Column(Integer, primary_key=True) name = Column(String(50)) - addresses = relationship('Address', - primaryjoin='User.id==Address.user_id', - backref='user') + addresses = relationship( + "Address", + primaryjoin="User.id==Address.user_id", + backref="user", + ) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" id = Column(Integer, primary_key=True) email = Column(String(50)) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) configure_mappers() - eq_(str(User.addresses.property.primaryjoin), - str(Address.user.property.primaryjoin)) + eq_( + str(User.addresses.property.primaryjoin), + str(Address.user.property.primaryjoin), + ) def test_string_dependency_resolution_tables(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' + __tablename__ = "users" id = Column(Integer, primary_key=True) name = Column(String(50)) - props = relationship('Prop', secondary='user_to_prop', - primaryjoin='User.id==user_to_prop.c.u' - 'ser_id', - secondaryjoin='user_to_prop.c.prop_id=' - '=Prop.id', backref='users') + props = relationship( + "Prop", + secondary="user_to_prop", + primaryjoin="User.id==user_to_prop.c.u" "ser_id", + secondaryjoin="user_to_prop.c.prop_id=" "=Prop.id", + backref="users", + ) class Prop(Base, fixtures.ComparableEntity): - __tablename__ = 'props' + __tablename__ = "props" id = Column(Integer, primary_key=True) name = Column(String(50)) user_to_prop = Table( - 'user_to_prop', Base.metadata, - Column('user_id', Integer, ForeignKey('users.id')), - Column('prop_id', Integer, ForeignKey('props.id'))) + "user_to_prop", + Base.metadata, + Column("user_id", Integer, ForeignKey("users.id")), + Column("prop_id", Integer, ForeignKey("props.id")), + ) configure_mappers() - assert class_mapper(User).get_property('props').secondary \ - is user_to_prop + assert ( + class_mapper(User).get_property("props").secondary is user_to_prop + ) def test_string_dependency_resolution_schemas(self): Base = decl.declarative_base() class User(Base): - __tablename__ = 'users' - __table_args__ = {'schema': 'fooschema'} + __tablename__ = "users" + __table_args__ = {"schema": "fooschema"} id = Column(Integer, primary_key=True) name = Column(String(50)) props = relationship( - 'Prop', secondary='fooschema.user_to_prop', - primaryjoin='User.id==fooschema.user_to_prop.c.user_id', - secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id', - backref='users') + "Prop", + secondary="fooschema.user_to_prop", + primaryjoin="User.id==fooschema.user_to_prop.c.user_id", + secondaryjoin="fooschema.user_to_prop.c.prop_id==Prop.id", + backref="users", + ) class Prop(Base): - __tablename__ = 'props' - __table_args__ = {'schema': 'fooschema'} + __tablename__ = "props" + __table_args__ = {"schema": "fooschema"} id = Column(Integer, primary_key=True) name = Column(String(50)) user_to_prop = Table( - 'user_to_prop', Base.metadata, - Column('user_id', Integer, ForeignKey('fooschema.users.id')), - Column('prop_id', Integer, ForeignKey('fooschema.props.id')), - schema='fooschema') + "user_to_prop", + Base.metadata, + Column("user_id", Integer, ForeignKey("fooschema.users.id")), + Column("prop_id", Integer, ForeignKey("fooschema.props.id")), + schema="fooschema", + ) configure_mappers() - assert class_mapper(User).get_property('props').secondary \ - is user_to_prop + assert ( + class_mapper(User).get_property("props").secondary is user_to_prop + ) def test_string_dependency_resolution_annotations(self): Base = decl.declarative_base() class Parent(Base): - __tablename__ = 'parent' + __tablename__ = "parent" id = Column(Integer, primary_key=True) name = Column(String) children = relationship( "Child", primaryjoin="Parent.name==" - "remote(foreign(func.lower(Child.name_upper)))" + "remote(foreign(func.lower(Child.name_upper)))", ) class Child(Base): - __tablename__ = 'child' + __tablename__ = "child" id = Column(Integer, primary_key=True) name_upper = Column(String) configure_mappers() eq_( Parent.children.property._calculated_foreign_keys, - set([Child.name_upper.property.columns[0]]) + set([Child.name_upper.property.columns[0]]), ) def test_shared_class_registry(self): @@ -787,11 +875,11 @@ class DeclarativeTest(DeclarativeTestBase): Base2 = decl.declarative_base(testing.db, class_registry=reg) class A(Base1): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) class B(Base2): - __tablename__ = 'b' + __tablename__ = "b" id = Column(Integer, primary_key=True) aid = Column(Integer, ForeignKey(A.id)) as_ = relationship("A") @@ -799,24 +887,28 @@ class DeclarativeTest(DeclarativeTestBase): assert B.as_.property.mapper.class_ is A def test_uncompiled_attributes_in_relationship(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) email = Column(String(50)) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) - addresses = relationship('Address', order_by=Address.email, - foreign_keys=Address.user_id, - remote_side=Address.user_id) + addresses = relationship( + "Address", + order_by=Address.email, + foreign_keys=Address.user_id, + remote_side=Address.user_id, + ) # get the mapper for User. User mapper will compile, # "addresses" relationship will call upon Address.user_id for @@ -829,28 +921,39 @@ class DeclarativeTest(DeclarativeTestBase): class_mapper(User) Base.metadata.create_all() sess = create_session() - u1 = User(name='ed', addresses=[ - Address(email='abc'), - Address(email='xyz'), Address(email='def')]) + u1 = User( + name="ed", + addresses=[ + Address(email="abc"), + Address(email="xyz"), + Address(email="def"), + ], + ) sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).filter(User.name == 'ed').one(), - User(name='ed', addresses=[ - Address(email='abc'), - Address(email='def'), Address(email='xyz')])) + eq_( + sess.query(User).filter(User.name == "ed").one(), + User( + name="ed", + addresses=[ + Address(email="abc"), + Address(email="def"), + Address(email="xyz"), + ], + ), + ) def test_nice_dependency_error(self): - class User(Base): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True) - addresses = relationship('Address') + __tablename__ = "users" + id = Column("id", Integer, primary_key=True) + addresses = relationship("Address") class Address(Base): - __tablename__ = 'addresses' + __tablename__ = "addresses" id = Column(Integer, primary_key=True) foo = sa.orm.column_property(User.id == 5) @@ -860,16 +963,15 @@ class DeclarativeTest(DeclarativeTestBase): assert_raises(sa.exc.ArgumentError, configure_mappers) def test_nice_dependency_error_works_with_hasattr(self): - class User(Base): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True) - addresses = relationship('Address') + __tablename__ = "users" + id = Column("id", Integer, primary_key=True) + addresses = relationship("Address") # hasattr() on a compile-loaded attribute try: - hasattr(User.addresses, 'property') + hasattr(User.addresses, "property") except exc.InvalidRequestError: assert sa.util.compat.py32 @@ -882,15 +984,16 @@ class DeclarativeTest(DeclarativeTestBase): " - can't proceed with initialization of other mappers. " r"Triggering mapper: 'Mapper\|User\|users'. " "Original exception was: When initializing.*", - configure_mappers) + configure_mappers, + ) def test_custom_base(self): class MyBase(object): - def foobar(self): return "foobar" + Base = decl.declarative_base(cls=MyBase) - assert hasattr(Base, 'metadata') + assert hasattr(Base, "metadata") assert Base().foobar() == "foobar" def test_uses_get_on_class_col_fk(self): @@ -899,22 +1002,23 @@ class DeclarativeTest(DeclarativeTestBase): class Master(Base): - __tablename__ = 'master' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "master" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) class Detail(Base): - __tablename__ = 'detail' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "detail" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) master_id = Column(None, ForeignKey(Master.id)) master = relationship(Master) Base.metadata.create_all() configure_mappers() - assert class_mapper(Detail).get_property('master' - ).strategy.use_get + assert class_mapper(Detail).get_property("master").strategy.use_get m1 = Master() d1 = Detail(master=m1) sess = create_session() @@ -931,12 +1035,12 @@ class DeclarativeTest(DeclarativeTestBase): def test_index_doesnt_compile(self): class User(Base): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) error = relationship("Address") - i = Index('my_index', User.name) + i = Index("my_index", User.name) # compile fails due to the nonexistent Addresses relationship assert_raises(sa.exc.InvalidRequestError, configure_mappers) @@ -950,83 +1054,96 @@ class DeclarativeTest(DeclarativeTestBase): Base.metadata.create_all() def test_add_prop(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) - User.name = Column('name', String(50)) - User.addresses = relationship('Address', backref='user') + User.name = Column("name", String(50)) + User.addresses = relationship("Address", backref="user") class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) - Address.email = Column(String(50), key='_email') - Address.user_id = Column('user_id', Integer, - ForeignKey('users.id'), key='_user_id') + Address.email = Column(String(50), key="_email") + Address.user_id = Column( + "user_id", Integer, ForeignKey("users.id"), key="_user_id" + ) Base.metadata.create_all() - eq_(Address.__table__.c['id'].name, 'id') - eq_(Address.__table__.c['_email'].name, 'email') - eq_(Address.__table__.c['_user_id'].name, 'user_id') - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + eq_(Address.__table__.c["id"].name, "id") + eq_(Address.__table__.c["_email"].name, "email") + eq_(Address.__table__.c["_user_id"].name, "user_id") + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User( - name='u1', - addresses=[Address(email='one'), Address(email='two')])]) - a1 = sess.query(Address).filter(Address.email == 'two').one() - eq_(a1, Address(email='two')) - eq_(a1.user, User(name='u1')) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) + a1 = sess.query(Address).filter(Address.email == "two").one() + eq_(a1, Address(email="two")) + eq_(a1.user, User(name="u1")) def test_alt_name_attr_subclass_column_inline(self): # [ticket:2900] class A(Base): - __tablename__ = 'a' - id = Column('id', Integer, primary_key=True) - data = Column('data') + __tablename__ = "a" + id = Column("id", Integer, primary_key=True) + data = Column("data") class ASub(A): brap = A.data + assert ASub.brap.property is A.data.property assert isinstance( - ASub.brap.original_property, properties.SynonymProperty) + ASub.brap.original_property, properties.SynonymProperty + ) def test_alt_name_attr_subclass_relationship_inline(self): # [ticket:2900] class A(Base): - __tablename__ = 'a' - id = Column('id', Integer, primary_key=True) - b_id = Column(Integer, ForeignKey('b.id')) + __tablename__ = "a" + id = Column("id", Integer, primary_key=True) + b_id = Column(Integer, ForeignKey("b.id")) b = relationship("B", backref="as_") class B(Base): - __tablename__ = 'b' - id = Column('id', Integer, primary_key=True) + __tablename__ = "b" + id = Column("id", Integer, primary_key=True) configure_mappers() class ASub(A): brap = A.b + assert ASub.brap.property is A.b.property assert isinstance( - ASub.brap.original_property, properties.SynonymProperty) + ASub.brap.original_property, properties.SynonymProperty + ) ASub(brap=B()) def test_alt_name_attr_subclass_column_attrset(self): # [ticket:2900] class A(Base): - __tablename__ = 'a' - id = Column('id', Integer, primary_key=True) - data = Column('data') + __tablename__ = "a" + id = Column("id", Integer, primary_key=True) + data = Column("data") + A.brap = A.data assert A.brap.property is A.data.property assert isinstance(A.brap.original_property, properties.SynonymProperty) @@ -1034,113 +1151,131 @@ class DeclarativeTest(DeclarativeTestBase): def test_alt_name_attr_subclass_relationship_attrset(self): # [ticket:2900] class A(Base): - __tablename__ = 'a' - id = Column('id', Integer, primary_key=True) - b_id = Column(Integer, ForeignKey('b.id')) + __tablename__ = "a" + id = Column("id", Integer, primary_key=True) + b_id = Column(Integer, ForeignKey("b.id")) b = relationship("B", backref="as_") + A.brap = A.b class B(Base): - __tablename__ = 'b' - id = Column('id', Integer, primary_key=True) + __tablename__ = "b" + id = Column("id", Integer, primary_key=True) assert A.brap.property is A.b.property assert isinstance(A.brap.original_property, properties.SynonymProperty) A(brap=B()) def test_eager_order_by(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', order_by=Address.email) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship("Address", order_by=Address.email) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='two'), - Address(email='one')]) + u1 = User( + name="u1", addresses=[Address(email="two"), Address(email="one")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).options(joinedload(User.addresses)).all(), - [User(name='u1', addresses=[Address(email='one'), - Address(email='two')])]) + eq_( + sess.query(User).options(joinedload(User.addresses)).all(), + [ + User( + name="u1", + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_order_by_multi(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', - order_by=(Address.email, Address.id)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship( + "Address", order_by=(Address.email, Address.id) + ) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='two'), - Address(email='one')]) + u1 = User( + name="u1", addresses=[Address(email="two"), Address(email="one")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - u = sess.query(User).filter(User.name == 'u1').one() + u = sess.query(User).filter(User.name == "u1").one() u.addresses def test_as_declarative(self): - class User(fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', backref='user') + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship("Address", backref="user") class Address(fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) reg = {} decl.instrument_declarative(User, reg, Base.metadata) decl.instrument_declarative(Address, reg, Base.metadata) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User( - name='u1', - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_custom_mapper_attribute(self): - def mymapper(cls, tbl, **kwargs): m = sa.orm.mapper(cls, tbl, **kwargs) m.CHECK = True @@ -1149,14 +1284,13 @@ class DeclarativeTest(DeclarativeTestBase): base = decl.declarative_base() class Foo(base): - __tablename__ = 'foo' + __tablename__ = "foo" __mapper_cls__ = mymapper id = Column(Integer, primary_key=True) eq_(Foo.__mapper__.CHECK, True) def test_custom_mapper_argument(self): - def mymapper(cls, tbl, **kwargs): m = sa.orm.mapper(cls, tbl, **kwargs) m.CHECK = True @@ -1165,7 +1299,7 @@ class DeclarativeTest(DeclarativeTestBase): base = decl.declarative_base(mapper=mymapper) class Foo(base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) eq_(Foo.__mapper__.CHECK, True) @@ -1173,23 +1307,22 @@ class DeclarativeTest(DeclarativeTestBase): def test_oops(self): with testing.expect_warnings( - "Ignoring declarative-like tuple value of " - "attribute 'name'"): + "Ignoring declarative-like tuple value of " "attribute 'name'" + ): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)), + __tablename__ = "users" + id = Column("id", Integer, primary_key=True) + name = (Column("name", String(50)),) def test_table_args_no_dict(self): - class Foo1(Base): - __tablename__ = 'foo' - __table_args__ = ForeignKeyConstraint(['id'], ['foo.bar']), - id = Column('id', Integer, primary_key=True) - bar = Column('bar', Integer) + __tablename__ = "foo" + __table_args__ = (ForeignKeyConstraint(["id"], ["foo.bar"]),) + id = Column("id", Integer, primary_key=True) + bar = Column("bar", Integer) assert Foo1.__table__.c.id.references(Foo1.__table__.c.bar) @@ -1197,49 +1330,50 @@ class DeclarativeTest(DeclarativeTestBase): def err(): class Foo1(Base): - __tablename__ = 'foo' - __table_args__ = ForeignKeyConstraint(['id'], ['foo.id' - ]) - id = Column('id', Integer, primary_key=True) - assert_raises_message(sa.exc.ArgumentError, - '__table_args__ value must be a tuple, ', err) + __tablename__ = "foo" + __table_args__ = ForeignKeyConstraint(["id"], ["foo.id"]) + id = Column("id", Integer, primary_key=True) - def test_table_args_none(self): + assert_raises_message( + sa.exc.ArgumentError, "__table_args__ value must be a tuple, ", err + ) + def test_table_args_none(self): class Foo2(Base): - __tablename__ = 'foo' + __tablename__ = "foo" __table_args__ = None - id = Column('id', Integer, primary_key=True) + id = Column("id", Integer, primary_key=True) assert Foo2.__table__.kwargs == {} def test_table_args_dict_format(self): - class Foo2(Base): - __tablename__ = 'foo' - __table_args__ = {'mysql_engine': 'InnoDB'} - id = Column('id', Integer, primary_key=True) + __tablename__ = "foo" + __table_args__ = {"mysql_engine": "InnoDB"} + id = Column("id", Integer, primary_key=True) - assert Foo2.__table__.kwargs['mysql_engine'] == 'InnoDB' + assert Foo2.__table__.kwargs["mysql_engine"] == "InnoDB" def test_table_args_tuple_format(self): class Foo2(Base): - __tablename__ = 'foo' - __table_args__ = {'mysql_engine': 'InnoDB'} - id = Column('id', Integer, primary_key=True) + __tablename__ = "foo" + __table_args__ = {"mysql_engine": "InnoDB"} + id = Column("id", Integer, primary_key=True) class Bar(Base): - __tablename__ = 'bar' - __table_args__ = ForeignKeyConstraint(['id'], ['foo.id']), \ - {'mysql_engine': 'InnoDB'} - id = Column('id', Integer, primary_key=True) + __tablename__ = "bar" + __table_args__ = ( + ForeignKeyConstraint(["id"], ["foo.id"]), + {"mysql_engine": "InnoDB"}, + ) + id = Column("id", Integer, primary_key=True) assert Bar.__table__.c.id.references(Foo2.__table__.c.id) - assert Bar.__table__.kwargs['mysql_engine'] == 'InnoDB' + assert Bar.__table__.kwargs["mysql_engine"] == "InnoDB" def test_table_cls_attribute(self): class Foo(Base): @@ -1248,7 +1382,7 @@ class DeclarativeTest(DeclarativeTestBase): @classmethod def __table_cls__(cls, *arg, **kw): name = arg[0] - return Table(name + 'bat', *arg[1:], **kw) + return Table(name + "bat", *arg[1:], **kw) id = Column(Integer, primary_key=True) @@ -1265,8 +1399,9 @@ class DeclarativeTest(DeclarativeTestBase): @classmethod def __table_cls__(cls, *arg, **kw): for obj in arg[1:]: - if (isinstance(obj, Column) and obj.primary_key) or \ - isinstance(obj, PrimaryKeyConstraint): + if ( + isinstance(obj, Column) and obj.primary_key + ) or isinstance(obj, PrimaryKeyConstraint): return Table(*arg, **kw) return None @@ -1280,77 +1415,99 @@ class DeclarativeTest(DeclarativeTestBase): is_(inspect(Employee).local_table, Person.__table__) def test_expression(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', backref='user') + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship("Address", backref="user") class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) - User.address_count = \ - sa.orm.column_property(sa.select([sa.func.count(Address.id)]). - where(Address.user_id - == User.id).as_scalar()) + User.address_count = sa.orm.column_property( + sa.select([sa.func.count(Address.id)]) + .where(Address.user_id == User.id) + .as_scalar() + ) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User(name='u1', address_count=2, - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + address_count=2, + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_useless_declared_attr(self): class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', backref='user') + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship("Address", backref="user") @declared_attr def address_count(cls): # this doesn't really gain us anything. but if # one is used, lets have it function as expected... return sa.orm.column_property( - sa.select([sa.func.count(Address.id)]). - where(Address.user_id == cls.id)) + sa.select([sa.func.count(Address.id)]).where( + Address.user_id == cls.id + ) + ) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User(name='u1', address_count=2, - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + address_count=2, + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_declared_on_base_class(self): class MyBase(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) @declared_attr @@ -1358,25 +1515,26 @@ class DeclarativeTest(DeclarativeTestBase): return Column(Integer) class MyClass(MyBase): - __tablename__ = 'bar' - id = Column(Integer, ForeignKey('foo.id'), primary_key=True) + __tablename__ = "bar" + id = Column(Integer, ForeignKey("foo.id"), primary_key=True) # previously, the 'somecol' declared_attr would be ignored # by the mapping and would remain unused. now we take # it as part of MyBase. - assert 'somecol' in MyBase.__table__.c - assert 'somecol' not in MyClass.__table__.c + assert "somecol" in MyBase.__table__.c + assert "somecol" not in MyClass.__table__.c def test_decl_cascading_warns_non_mixin(self): with expect_warnings( - "Use of @declared_attr.cascading only applies to " - "Declarative 'mixin' and 'abstract' classes. " - "Currently, this flag is ignored on mapped class " - "<class '.*.MyBase'>" + "Use of @declared_attr.cascading only applies to " + "Declarative 'mixin' and 'abstract' classes. " + "Currently, this flag is ignored on mapped class " + "<class '.*.MyBase'>" ): + class MyBase(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) @declared_attr.cascading @@ -1384,109 +1542,116 @@ class DeclarativeTest(DeclarativeTestBase): return Column(Integer) def test_column(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) - User.a = Column('a', String(10)) + User.a = Column("a", String(10)) User.b = Column(String(10)) Base.metadata.create_all() - u1 = User(name='u1', a='a', b='b') - eq_(u1.a, 'a') - eq_(User.a.get_history(u1), (['a'], (), ())) + u1 = User(name="u1", a="a", b="b") + eq_(u1.a, "a") + eq_(User.a.get_history(u1), (["a"], (), ())) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [User(name='u1', a='a', b='b')]) + eq_(sess.query(User).all(), [User(name="u1", a="a", b="b")]) def test_column_properties(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) email = Column(String(50)) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) - adr_count = \ - sa.orm.column_property( - sa.select([sa.func.count(Address.id)], - Address.user_id == id).as_scalar()) + adr_count = sa.orm.column_property( + sa.select( + [sa.func.count(Address.id)], Address.user_id == id + ).as_scalar() + ) addresses = relationship(Address) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User(name='u1', adr_count=2, - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + adr_count=2, + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_column_properties_2(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' + __tablename__ = "addresses" id = Column(Integer, primary_key=True) email = Column(String(50)) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column("id", Integer, primary_key=True) + name = Column("name", String(50)) # this is not "valid" but we want to test that Address.id # doesn't get stuck into user's table adr_count = Address.id - eq_(set(User.__table__.c.keys()), set(['id', 'name'])) - eq_(set(Address.__table__.c.keys()), set(['id', 'email', - 'user_id'])) + eq_(set(User.__table__.c.keys()), set(["id", "name"])) + eq_(set(Address.__table__.c.keys()), set(["id", "email", "user_id"])) def test_deferred(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = sa.orm.deferred(Column(String(50))) Base.metadata.create_all() sess = create_session() - sess.add(User(name='u1')) + sess.add(User(name="u1")) sess.flush() sess.expunge_all() - u1 = sess.query(User).filter(User.name == 'u1').one() - assert 'name' not in u1.__dict__ + u1 = sess.query(User).filter(User.name == "u1").one() + assert "name" not in u1.__dict__ def go(): - eq_(u1.name, 'u1') + eq_(u1.name, "u1") self.assert_sql_count(testing.db, go, 1) def test_composite_inline(self): class AddressComposite(fixtures.ComparableEntity): - def __init__(self, street, state): self.street = street self.state = state @@ -1495,30 +1660,27 @@ class DeclarativeTest(DeclarativeTestBase): return [self.street, self.state] class User(Base, fixtures.ComparableEntity): - __tablename__ = 'user' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) - address = composite(AddressComposite, - Column('street', String(50)), - Column('state', String(2)), - ) + __tablename__ = "user" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) + address = composite( + AddressComposite, + Column("street", String(50)), + Column("state", String(2)), + ) Base.metadata.create_all() sess = Session() - sess.add(User( - address=AddressComposite('123 anywhere street', - 'MD') - )) + sess.add(User(address=AddressComposite("123 anywhere street", "MD"))) sess.commit() eq_( sess.query(User).all(), - [User(address=AddressComposite('123 anywhere street', - 'MD'))] + [User(address=AddressComposite("123 anywhere street", "MD"))], ) def test_composite_separate(self): class AddressComposite(fixtures.ComparableEntity): - def __init__(self, street, state): self.street = street self.state = state @@ -1527,78 +1689,79 @@ class DeclarativeTest(DeclarativeTestBase): return [self.street, self.state] class User(Base, fixtures.ComparableEntity): - __tablename__ = 'user' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "user" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) street = Column(String(50)) state = Column(String(2)) - address = composite(AddressComposite, - street, state) + address = composite(AddressComposite, street, state) Base.metadata.create_all() sess = Session() - sess.add(User( - address=AddressComposite('123 anywhere street', - 'MD') - )) + sess.add(User(address=AddressComposite("123 anywhere street", "MD"))) sess.commit() eq_( sess.query(User).all(), - [User(address=AddressComposite('123 anywhere street', - 'MD'))] + [User(address=AddressComposite("123 anywhere street", "MD"))], ) def test_mapping_to_join(self): - users = Table('users', Base.metadata, - Column('id', Integer, primary_key=True) - ) - addresses = Table('addresses', Base.metadata, - Column('id', Integer, primary_key=True), - Column('user_id', Integer, ForeignKey('users.id')) - ) - usersaddresses = sa.join(users, addresses, users.c.id - == addresses.c.user_id) + users = Table( + "users", Base.metadata, Column("id", Integer, primary_key=True) + ) + addresses = Table( + "addresses", + Base.metadata, + Column("id", Integer, primary_key=True), + Column("user_id", Integer, ForeignKey("users.id")), + ) + usersaddresses = sa.join( + users, addresses, users.c.id == addresses.c.user_id + ) class User(Base): __table__ = usersaddresses - __table_args__ = {'primary_key': [users.c.id]} + __table_args__ = {"primary_key": [users.c.id]} # need to use column_property for now user_id = column_property(users.c.id, addresses.c.user_id) address_id = addresses.c.id - assert User.__mapper__.get_property('user_id').columns[0] \ - is users.c.id - assert User.__mapper__.get_property('user_id').columns[1] \ + assert User.__mapper__.get_property("user_id").columns[0] is users.c.id + assert ( + User.__mapper__.get_property("user_id").columns[1] is addresses.c.user_id + ) def test_synonym_inline(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - _name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + _name = Column("name", String(50)) def _set_name(self, name): - self._name = 'SOMENAME ' + name + self._name = "SOMENAME " + name def _get_name(self): return self._name - name = sa.orm.synonym('_name', - descriptor=property(_get_name, - _set_name)) + name = sa.orm.synonym( + "_name", descriptor=property(_get_name, _set_name) + ) Base.metadata.create_all() sess = create_session() - u1 = User(name='someuser') - eq_(u1.name, 'SOMENAME someuser') + u1 = User(name="someuser") + eq_(u1.name, "SOMENAME someuser") sess.add(u1) sess.flush() - eq_(sess.query(User).filter(User.name == 'SOMENAME someuser' - ).one(), u1) + eq_( + sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1 + ) def test_synonym_no_descriptor(self): from sqlalchemy.orm.properties import ColumnProperty @@ -1608,68 +1771,70 @@ class DeclarativeTest(DeclarativeTestBase): __hash__ = None def __eq__(self, other): - return self.__clause_element__() == other + ' FOO' + return self.__clause_element__() == other + " FOO" class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - _name = Column('name', String(50)) - name = sa.orm.synonym('_name', - comparator_factory=CustomCompare) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + _name = Column("name", String(50)) + name = sa.orm.synonym("_name", comparator_factory=CustomCompare) Base.metadata.create_all() sess = create_session() - u1 = User(name='someuser FOO') + u1 = User(name="someuser FOO") sess.add(u1) sess.flush() - eq_(sess.query(User).filter(User.name == 'someuser').one(), u1) + eq_(sess.query(User).filter(User.name == "someuser").one(), u1) def test_synonym_added(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - _name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + _name = Column("name", String(50)) def _set_name(self, name): - self._name = 'SOMENAME ' + name + self._name = "SOMENAME " + name def _get_name(self): return self._name name = property(_get_name, _set_name) - User.name = sa.orm.synonym('_name', descriptor=User.name) + User.name = sa.orm.synonym("_name", descriptor=User.name) Base.metadata.create_all() sess = create_session() - u1 = User(name='someuser') - eq_(u1.name, 'SOMENAME someuser') + u1 = User(name="someuser") + eq_(u1.name, "SOMENAME someuser") sess.add(u1) sess.flush() - eq_(sess.query(User).filter(User.name == 'SOMENAME someuser' - ).one(), u1) + eq_( + sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1 + ) def test_reentrant_compile_via_foreignkey(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', backref='user') + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship("Address", backref="user") class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey(User.id)) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey(User.id)) # previous versions would force a re-entrant mapper compile via # the User.id inside the ForeignKey but this is no longer the @@ -1678,65 +1843,83 @@ class DeclarativeTest(DeclarativeTestBase): sa.orm.configure_mappers() eq_( list(Address.user_id.property.columns[0].foreign_keys)[0].column, - User.__table__.c.id + User.__table__.c.id, ) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User(name='u1', - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_relationship_reference(self): - class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - email = Column('email', String(50)) - user_id = Column('user_id', Integer, ForeignKey('users.id')) + __tablename__ = "addresses" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + email = Column("email", String(50)) + user_id = Column("user_id", Integer, ForeignKey("users.id")) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) - addresses = relationship('Address', backref='user', - primaryjoin=id == Address.user_id) - - User.address_count = \ - sa.orm.column_property(sa.select([sa.func.count(Address.id)]). - where(Address.user_id - == User.id).as_scalar()) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) + addresses = relationship( + "Address", backref="user", primaryjoin=id == Address.user_id + ) + + User.address_count = sa.orm.column_property( + sa.select([sa.func.count(Address.id)]) + .where(Address.user_id == User.id) + .as_scalar() + ) Base.metadata.create_all() - u1 = User(name='u1', addresses=[Address(email='one'), - Address(email='two')]) + u1 = User( + name="u1", addresses=[Address(email="one"), Address(email="two")] + ) sess = create_session() sess.add(u1) sess.flush() sess.expunge_all() - eq_(sess.query(User).all(), [ - User(name='u1', address_count=2, - addresses=[Address(email='one'), Address(email='two')])]) + eq_( + sess.query(User).all(), + [ + User( + name="u1", + address_count=2, + addresses=[Address(email="one"), Address(email="two")], + ) + ], + ) def test_pk_with_fk_init(self): - class Bar(Base): - __tablename__ = 'bar' - id = sa.Column(sa.Integer, sa.ForeignKey('foo.id'), - primary_key=True) + __tablename__ = "bar" + id = sa.Column( + sa.Integer, sa.ForeignKey("foo.id"), primary_key=True + ) ex = sa.Column(sa.Integer, primary_key=True) class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = sa.Column(sa.Integer, primary_key=True) bars = sa.orm.relationship(Bar) @@ -1746,108 +1929,101 @@ class DeclarativeTest(DeclarativeTestBase): def test_with_explicit_autoloaded(self): meta = MetaData(testing.db) t1 = Table( - 't1', meta, - Column('id', String(50), primary_key=True), - Column('data', String(50))) + "t1", + meta, + Column("id", String(50), primary_key=True), + Column("data", String(50)), + ) meta.create_all() try: class MyObj(Base): - __table__ = Table('t1', Base.metadata, autoload=True) + __table__ = Table("t1", Base.metadata, autoload=True) sess = create_session() - m = MyObj(id='someid', data='somedata') + m = MyObj(id="someid", data="somedata") sess.add(m) sess.flush() - eq_(t1.select().execute().fetchall(), [('someid', 'somedata' - )]) + eq_(t1.select().execute().fetchall(), [("someid", "somedata")]) finally: meta.drop_all() def test_synonym_for(self): - class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) - @decl.synonym_for('name') + @decl.synonym_for("name") @property def namesyn(self): return self.name Base.metadata.create_all() sess = create_session() - u1 = User(name='someuser') - eq_(u1.name, 'someuser') - eq_(u1.namesyn, 'someuser') + u1 = User(name="someuser") + eq_(u1.name, "someuser") + eq_(u1.namesyn, "someuser") sess.add(u1) sess.flush() - rt = sess.query(User).filter(User.namesyn == 'someuser').one() + rt = sess.query(User).filter(User.namesyn == "someuser").one() eq_(rt, u1) def test_comparable_using(self): - class NameComparator(sa.orm.PropComparator): - @property def upperself(self): cls = self.prop.parent.class_ - col = getattr(cls, 'name') + col = getattr(cls, "name") return sa.func.upper(col) - def operate( - self, - op, - other, - **kw - ): + def operate(self, op, other, **kw): return op(self.upperself, other, **kw) class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column('id', Integer, primary_key=True, - test_needs_autoincrement=True) - name = Column('name', String(50)) + __tablename__ = "users" + id = Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ) + name = Column("name", String(50)) @decl.comparable_using(NameComparator) @property def uc_name(self): - return self.name is not None and self.name.upper() \ - or None + return self.name is not None and self.name.upper() or None Base.metadata.create_all() sess = create_session() - u1 = User(name='someuser') - eq_(u1.name, 'someuser', u1.name) - eq_(u1.uc_name, 'SOMEUSER', u1.uc_name) + u1 = User(name="someuser") + eq_(u1.name, "someuser", u1.name) + eq_(u1.uc_name, "SOMEUSER", u1.uc_name) sess.add(u1) sess.flush() sess.expunge_all() - rt = sess.query(User).filter(User.uc_name == 'SOMEUSER').one() + rt = sess.query(User).filter(User.uc_name == "SOMEUSER").one() eq_(rt, u1) sess.expunge_all() - rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE' - )).one() + rt = sess.query(User).filter(User.uc_name.startswith("SOMEUSE")).one() eq_(rt, u1) def test_duplicate_classes_in_base(self): - class Test(Base): - __tablename__ = 'a' + __tablename__ = "a" id = Column(Integer, primary_key=True) assert_raises_message( sa.exc.SAWarning, "This declarative base already contains a class with ", - lambda: type(Base)("Test", (Base,), dict( - __tablename__='b', - id=Column(Integer, primary_key=True) - )) + lambda: type(Base)( + "Test", + (Base,), + dict(__tablename__="b", id=Column(Integer, primary_key=True)), + ), ) @testing.teardown_events(MapperEvents) @@ -1865,19 +2041,18 @@ class DeclarativeTest(DeclarativeTestBase): canary.class_instrument(cls) class Test(Base): - __tablename__ = 'test' + __tablename__ = "test" id = Column(Integer, primary_key=True) eq_( canary.mock_calls, [ mock.call.instrument_class(Test.__mapper__, Test), - mock.call.class_instrument(Test) - ] + mock.call.class_instrument(Test), + ], ) def test_cls_docstring(self): - class MyBase(object): """MyBase Docstring""" @@ -1889,7 +2064,7 @@ class DeclarativeTest(DeclarativeTestBase): Base = decl.declarative_base() class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) data = Column(String) @@ -1900,14 +2075,14 @@ class DeclarativeTest(DeclarativeTestBase): assert_raises_message( NotImplementedError, "Can't un-map individual mapped attributes on a mapped class.", - go + go, ) def test_delattr_hybrid_fine(self): Base = decl.declarative_base() class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) data = Column(String) @@ -1928,7 +2103,7 @@ class DeclarativeTest(DeclarativeTestBase): Base = decl.declarative_base() class Foo(Base): - __tablename__ = 'foo' + __tablename__ = "foo" id = Column(Integer, primary_key=True) data = Column(String) @@ -1938,6 +2113,7 @@ class DeclarativeTest(DeclarativeTestBase): @hybrid_property def data_hybrid(self): return self.data + Foo.data_hybrid = data_hybrid assert "data_hybrid" in Foo.__mapper__.all_orm_descriptors.keys() @@ -1949,9 +2125,7 @@ class DeclarativeTest(DeclarativeTestBase): def _produce_test(inline, stringbased): - class ExplicitJoinTest(fixtures.MappedTest): - @classmethod def define_tables(cls, metadata): global User, Address @@ -1959,57 +2133,74 @@ def _produce_test(inline, stringbased): class User(Base, fixtures.ComparableEntity): - __tablename__ = 'users' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "users" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) name = Column(String(50)) class Address(Base, fixtures.ComparableEntity): - __tablename__ = 'addresses' - id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + __tablename__ = "addresses" + id = Column( + Integer, primary_key=True, test_needs_autoincrement=True + ) email = Column(String(50)) - user_id = Column(Integer, ForeignKey('users.id')) + user_id = Column(Integer, ForeignKey("users.id")) if inline: if stringbased: user = relationship( - 'User', - primaryjoin='User.id==Address.user_id', - backref='addresses') + "User", + primaryjoin="User.id==Address.user_id", + backref="addresses", + ) else: - user = relationship(User, primaryjoin=User.id - == user_id, backref='addresses') + user = relationship( + User, + primaryjoin=User.id == user_id, + backref="addresses", + ) if not inline: configure_mappers() if stringbased: Address.user = relationship( - 'User', - primaryjoin='User.id==Address.user_id', - backref='addresses') + "User", + primaryjoin="User.id==Address.user_id", + backref="addresses", + ) else: Address.user = relationship( User, primaryjoin=User.id == Address.user_id, - backref='addresses') + backref="addresses", + ) @classmethod def insert_data(cls): params = [ - dict(list(zip(('id', 'name'), column_values))) + dict(list(zip(("id", "name"), column_values))) for column_values in [ - (7, 'jack'), (8, 'ed'), - (9, 'fred'), (10, 'chuck')]] + (7, "jack"), + (8, "ed"), + (9, "fred"), + (10, "chuck"), + ] + ] User.__table__.insert().execute(params) - Address.__table__.insert().execute([ - dict(list(zip(('id', 'user_id', 'email'), column_values))) - for column_values in [ - (1, 7, 'jack@bean.com'), - (2, 8, 'ed@wood.com'), - (3, 8, 'ed@bettyboop.com'), - (4, 8, 'ed@lala.com'), (5, 9, 'fred@fred.com')]]) + Address.__table__.insert().execute( + [ + dict(list(zip(("id", "user_id", "email"), column_values))) + for column_values in [ + (1, 7, "jack@bean.com"), + (2, 8, "ed@wood.com"), + (3, 8, "ed@bettyboop.com"), + (4, 8, "ed@lala.com"), + (5, 9, "fred@fred.com"), + ] + ] + ) def test_aliased_join(self): @@ -2022,20 +2213,24 @@ def _produce_test(inline, stringbased): # _orm_adapt, though. sess = create_session() - eq_(sess.query(User).join(User.addresses, - aliased=True).filter( - Address.email == 'ed@wood.com').filter( - User.addresses.any(Address.email == 'jack@bean.com')).all(), - []) - - ExplicitJoinTest.__name__ = 'ExplicitJoinTest%s%s' % ( - inline and 'Inline' or 'Separate', - stringbased and 'String' or 'Literal') + eq_( + sess.query(User) + .join(User.addresses, aliased=True) + .filter(Address.email == "ed@wood.com") + .filter(User.addresses.any(Address.email == "jack@bean.com")) + .all(), + [], + ) + + ExplicitJoinTest.__name__ = "ExplicitJoinTest%s%s" % ( + inline and "Inline" or "Separate", + stringbased and "String" or "Literal", + ) return ExplicitJoinTest for inline in True, False: for stringbased in True, False: testclass = _produce_test(inline, stringbased) - exec('%s = testclass' % testclass.__name__) + exec("%s = testclass" % testclass.__name__) del testclass |