summaryrefslogtreecommitdiff
path: root/test/ext/declarative/test_basic.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 01:14:26 -0500
committermike bayer <mike_mp@zzzcomputing.com>2019-01-06 17:34:50 +0000
commit1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch)
tree28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/ext/declarative/test_basic.py
parent404e69426b05a82d905cbb3ad33adafccddb00dd (diff)
downloadsqlalchemy-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/ext/declarative/test_basic.py')
-rw-r--r--test/ext/declarative/test_basic.py1673
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