-from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message
+from sqlalchemy.test.testing import eq_, assert_raises, \
+ assert_raises_message
from sqlalchemy.ext import declarative as decl
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.test import testing
-from sqlalchemy import MetaData, Integer, String, ForeignKey, ForeignKeyConstraint, asc, Index
+from sqlalchemy import MetaData, Integer, String, ForeignKey, \
+ ForeignKeyConstraint, asc, Index
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import relationship, create_session, class_mapper, \
- joinedload, compile_mappers, backref, \
- clear_mappers, polymorphic_union, \
- deferred, column_property
-
+ joinedload, compile_mappers, backref, clear_mappers, \
+ polymorphic_union, deferred, column_property
from sqlalchemy.test.testing import eq_
from sqlalchemy.util import classproperty
-
from test.orm._base import ComparableEntity, MappedTest
class DeclarativeTestBase(testing.TestBase, testing.AssertsExecutionResults):
def go():
class User(Base):
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_cant_add_columns(self):
- t = Table('t', Base.metadata, Column('id', Integer, primary_key=True), Column('data', String))
+ t = Table('t', Base.metadata, Column('id', Integer,
+ primary_key=True), Column('data', String))
+
def go():
class User(Base):
__table__ = t
foo = Column(Integer, primary_key=True)
+
# 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
+
class Bar(Base):
__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_difficult_class(self):
"""test no getattr() errors with a customized class"""
def test_string_dependency_resolution(self):
from sqlalchemy.sql import desc
-
+
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ 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]")
- )
+ 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]'))
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ 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')])
+ u1 = User(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')])
- )
-
+ User(name='ed', addresses=[Address(email='xyz'),
+ Address(email='def'), Address(email='abc')]))
+
class Foo(Base, ComparableEntity):
+
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
- rel = relationship("User", primaryjoin="User.addresses==Foo.id")
- assert_raises_message(exc.InvalidRequestError, "'addresses' is not an instance of ColumnProperty", compile_mappers)
+ rel = relationship('User',
+ primaryjoin='User.addresses==Foo.id')
+
+ assert_raises_message(exc.InvalidRequestError,
+ "'addresses' is not an instance of "
+ "ColumnProperty", compile_mappers)
def test_string_dependency_resolution_two(self):
+
class User(Base, ComparableEntity):
__tablename__ = 'users'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column(String(50))
-
+
class Bar(Base, ComparableEntity):
__tablename__ = 'bar'
id = Column(Integer, primary_key=True)
- rel = relationship("User", primaryjoin="User.id==Bar.__table__.id")
- assert_raises_message(exc.InvalidRequestError, "does not have a mapped column named '__table__'", compile_mappers)
+ rel = relationship('User',
+ primaryjoin='User.id==Bar.__table__.id')
+
+ assert_raises_message(exc.InvalidRequestError,
+ "does not have a mapped column named "
+ "'__table__'", compile_mappers)
def test_string_dependency_resolution_no_magic(self):
"""test that full tinkery expressions work as written"""
-
+
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
- addresses = relationship("Address",
- primaryjoin="User.id==Address.user_id.prop.columns[0]")
-
+ addresses = relationship('Address',
+ primaryjoin='User.id==Address.user_id.prop.columns['
+ '0]')
+
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
-
+
compile_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, ComparableEntity):
+
__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, ComparableEntity):
+
__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'))
compile_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, ComparableEntity):
+
__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.user_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, ComparableEntity):
+
__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 = Table('user_to_prop', Base.metadata,
+ Column('user_id', Integer,
+ ForeignKey('users.id')), Column('prop_id',
+ Integer, ForeignKey('props.id')))
compile_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_uncompiled_attributes_in_relationship(self):
+
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ 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
# its clause element. Address.user_id is a _CompileOnAttr,
# which then calls class_mapper(Address). But ! We're already
- # "in compilation", but class_mapper(Address) needs to initialize
- # regardless, or COA's assertion fails
- # and things generally go downhill from there.
+ # "in compilation", but class_mapper(Address) needs to
+ # initialize regardless, or COA's assertion fails and things
+ # generally go downhill from there.
+
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')])
- )
+ 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")
+ addresses = relationship('Address')
class Address(Base):
- __tablename__ = 'addresses'
+ __tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
foo = sa.orm.column_property(User.id == 5)
- # this used to raise an error when accessing User.id but that's no longer the case
- # since we got rid of _CompileOnAttr.
+ # this used to raise an error when accessing User.id but that's
+ # no longer the case since we got rid of _CompileOnAttr.
+
assert_raises(sa.exc.ArgumentError, compile_mappers)
-
+
def test_nice_dependency_error_works_with_hasattr(self):
+
class User(Base):
+
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
- addresses = relationship("Addresss")
+ addresses = relationship('Addresss')
# hasattr() on a compile-loaded attribute
+
hasattr(User.addresses, 'property')
+
# the exeption is preserved
- assert_raises_message(sa.exc.InvalidRequestError,
- r"suppressed within a hasattr\(\)", compile_mappers)
+
+ assert_raises_message(sa.exc.InvalidRequestError,
+ r"suppressed within a hasattr\(\)",
+ compile_mappers)
def test_custom_base(self):
class MyBase(object):
assert Base().foobar() == "foobar"
def test_uses_get_on_class_col_fk(self):
+
# test [ticket:1492]
-
- class Master(Base):
- __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)
- master_id = Column(None, ForeignKey(Master.id))
- master = relationship(Master)
+ class Master(Base):
+
+ __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)
+ master_id = Column(None, ForeignKey(Master.id))
+ master = relationship(Master)
Base.metadata.create_all()
-
compile_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()
sess.add(d1)
sess.flush()
sess.expunge_all()
-
d1 = sess.query(Detail).first()
m1 = sess.query(Master).first()
+
def go():
assert d1.master
+
self.assert_sql_count(testing.db, go, 0)
def test_index_doesnt_compile(self):
Base.metadata.create_all()
def test_add_prop(self):
+
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
User.name = Column('name', String(50))
- User.addresses = relationship("Address", backref="user")
+ User.addresses = relationship('Address', backref='user')
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
- 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.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'),
- ])
+ 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'))
def test_eager_order_by(self):
+
class Address(Base, ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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, 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)
name = Column('name', String(50))
- addresses = relationship("Address", order_by=Address.email)
+ 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, ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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, 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)
name = Column('name', String(50))
- addresses = relationship("Address", order_by=(Address.email, Address.id))
+ 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()
a = u.addresses
-
+
def test_as_declarative(self):
+
class User(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)
name = Column('name', String(50))
- addresses = relationship("Address", backref="user")
+ addresses = relationship('Address', backref='user')
class Address(ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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'),
- ])])
-
@testing.uses_deprecated()
def test_custom_mapper(self):
+
class MyExt(sa.orm.MapperExtension):
+
def create_instance(self):
- return "CHECK"
+ return 'CHECK'
def mymapper(cls, tbl, **kwargs):
kwargs['extension'] = MyExt()
return sa.orm.mapper(cls, tbl, **kwargs)
from sqlalchemy.orm.mapper import Mapper
+
class MyMapper(Mapper):
+
def __init__(self, *args, **kwargs):
kwargs['extension'] = MyExt()
Mapper.__init__(self, *args, **kwargs)
ss = scoping.ScopedSession(create_session)
ss.extension = MyExt()
ss_mapper = ss.mapper
-
- for mapperfunc in (mymapper, MyMapper, ss_mapper):
+ for mapperfunc in mymapper, MyMapper, ss_mapper:
base = decl.declarative_base()
+
class Foo(base):
+
__tablename__ = 'foo'
__mapper_cls__ = mapperfunc
id = Column(Integer, primary_key=True)
- eq_(Foo.__mapper__.compile().extension.create_instance(), 'CHECK')
+ eq_(Foo.__mapper__.compile().extension.create_instance(),
+ 'CHECK')
base = decl.declarative_base(mapper=mapperfunc)
+
class Foo(base):
+
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
- eq_(Foo.__mapper__.compile().extension.create_instance(), 'CHECK')
+ eq_(Foo.__mapper__.compile().extension.create_instance(),
+ 'CHECK')
@testing.emits_warning('Ignoring declarative-like tuple value of '
'attribute id')
def test_oops(self):
+
def define():
+
class User(Base, ComparableEntity):
- __tablename__ = 'users'
+ __tablename__ = 'users'
id = Column('id', Integer, primary_key=True),
name = Column('name', String(50))
+
assert False
- assert_raises_message(
- sa.exc.ArgumentError,
- "Mapper Mapper|User|users could not assemble any primary key",
- define)
+
+ assert_raises_message(sa.exc.ArgumentError,
+ 'Mapper Mapper|User|users could not '
+ 'assemble any primary key', define)
def test_table_args(self):
-
+
def err():
class Foo(Base):
+
__tablename__ = 'foo'
- __table_args__ = (ForeignKeyConstraint(['id'], ['foo.id']),)
+ __table_args__ = ForeignKeyConstraint(['id'], ['foo.id'
+ ]),
id = Column('id', Integer, primary_key=True)
-
- assert_raises_message(sa.exc.ArgumentError, "Tuple form of __table_args__ is ", err)
-
+
+ assert_raises_message(sa.exc.ArgumentError,
+ 'Tuple form of __table_args__ is ', err)
+
class Foo(Base):
+
__tablename__ = 'foo'
- __table_args__ = {'mysql_engine':'InnoDB'}
+ __table_args__ = {'mysql_engine': 'InnoDB'}
id = Column('id', Integer, primary_key=True)
-
+
assert Foo.__table__.kwargs['mysql_engine'] == 'InnoDB'
class Bar(Base):
+
__tablename__ = 'bar'
- __table_args__ = (ForeignKeyConstraint(['id'], ['foo.id']), {'mysql_engine':'InnoDB'})
+ __table_args__ = ForeignKeyConstraint(['id'], ['foo.id']), \
+ {'mysql_engine': 'InnoDB'}
id = Column('id', Integer, primary_key=True)
-
+
assert Bar.__table__.c.id.references(Foo.__table__.c.id)
assert Bar.__table__.kwargs['mysql_engine'] == 'InnoDB'
-
+
def test_expression(self):
+
class User(Base, 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)
name = Column('name', String(50))
- addresses = relationship("Address", backref="user")
+ addresses = relationship('Address', backref='user')
class Address(Base, ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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_column(self):
+
class User(Base, 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)
name = Column('name', String(50))
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'], (), ()))
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, ComparableEntity):
+
__tablename__ = 'addresses'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
class User(Base, 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)
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, ComparableEntity):
+
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
class User(Base, ComparableEntity):
- __tablename__ = 'users'
+ __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 doesnt
- # get stuck into user's table
+
+ # this is not "valid" but we want to test that Address.id
+ # doesnt 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(Address.__table__.c.keys()), set(['id', 'email',
+ 'user_id']))
def test_deferred(self):
+
class User(Base, 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.add(User(name='u1'))
sess.flush()
sess.expunge_all()
-
u1 = sess.query(User).filter(User.name == 'u1').one()
assert 'name' not in u1.__dict__
+
def go():
eq_(u1.name, 'u1')
+
self.assert_sql_count(testing.db, go, 1)
def test_synonym_inline(self):
+
class User(Base, 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)
_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))
+ descriptor=property(_get_name,
+ _set_name))
Base.metadata.create_all()
-
sess = create_session()
u1 = User(name='someuser')
- eq_(u1.name, "SOMENAME 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
-
+
class CustomCompare(ColumnProperty.Comparator):
+
__hash__ = None
+
def __eq__(self, other):
return self.__clause_element__() == other + ' FOO'
-
+
class User(Base, 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)
_name = Column('name', String(50))
- name = sa.orm.synonym('_name', comparator_factory=CustomCompare)
-
- Base.metadata.create_all()
+ name = sa.orm.synonym('_name',
+ comparator_factory=CustomCompare)
+ Base.metadata.create_all()
sess = create_session()
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, 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)
_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")
+ 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, 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)
name = Column('name', String(50))
- addresses = relationship("Address", backref="user")
+ addresses = relationship('Address', backref='user')
class Address(Base, ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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 case
- sa.orm.compile_mappers()
+ # previous versions would force a re-entrant mapper compile via
+ # the User.id inside the ForeignKey but this is no longer the
+ # case
- eq_(str(Address.user_id.property.columns[0].foreign_keys[0]), "ForeignKey('users.id')")
-
+ sa.orm.compile_mappers()
+ eq_(str(Address.user_id.property.columns[0].foreign_keys[0]),
+ "ForeignKey('users.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, ComparableEntity):
- __tablename__ = 'addresses'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ __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, 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)
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())
+ 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)
-
+
assert Bar.__mapper__.primary_key[0] is Bar.__table__.c.id
assert Bar.__mapper__.primary_key[1] is Bar.__table__.c.ex
-
def test_with_explicit_autoloaded(self):
meta = MetaData(testing.db)
- t1 = Table('t1', meta,
- Column('id', String(50), primary_key=True, test_needs_autoincrement=True),
+ t1 = Table('t1', meta, Column('id', String(50),
+ primary_key=True, test_needs_autoincrement=True),
Column('data', String(50)))
meta.create_all()
try:
+
class MyObj(Base):
+
__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, 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)
name = Column('name', String(50))
@decl.synonym_for('name')
return self.name
Base.metadata.create_all()
-
sess = create_session()
u1 = User(name='someuser')
- eq_(u1.name, "someuser")
+ eq_(u1.name, 'someuser')
eq_(u1.namesyn, 'someuser')
sess.add(u1)
sess.flush()
-
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')
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, 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)
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.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()
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)
-
class DeclarativeInheritanceTest(DeclarativeTestBase):
-
+
def test_we_must_copy_mapper_args(self):
+
class Person(Base):
+
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on': discriminator,'polymorphic_identity':'person'}
-
+ __mapper_args__ = {'polymorphic_on': discriminator,
+ 'polymorphic_identity': 'person'}
+
class Engineer(Person):
+
primary_language = Column(String(50))
assert 'inherits' not in Person.__mapper_args__
assert class_mapper(Engineer).polymorphic_on is None
-
-
+
def test_custom_join_condition(self):
+
class Foo(Base):
+
__tablename__ = 'foo'
id = Column('id', Integer, primary_key=True)
class Bar(Foo):
+
__tablename__ = 'bar'
id = Column('id', Integer, primary_key=True)
foo_id = Column('foo_id', Integer)
- __mapper_args__ = {'inherit_condition':foo_id==Foo.id}
-
+ __mapper_args__ = {'inherit_condition': foo_id == Foo.id}
+
# compile succeeds because inherit_condition is honored
+
compile_mappers()
-
+
def test_joined(self):
+
class Company(Base, ComparableEntity):
+
__tablename__ = 'companies'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column('name', String(50))
- employees = relationship("Person")
+ employees = relationship('Person')
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
company_id = Column('company_id', Integer,
ForeignKey('companies.id'))
name = Column('name', String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
primary_language = Column('primary_language', String(50))
class Manager(Person):
+
__tablename__ = 'managers'
- __mapper_args__ = {'polymorphic_identity':'manager'}
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
golf_swing = Column('golf_swing', String(50))
Base.metadata.create_all()
-
sess = create_session()
-
- c1 = Company(name="MegaCorp, Inc.", employees=[
- Engineer(name="dilbert", primary_language="java"),
- Engineer(name="wally", primary_language="c++"),
- Manager(name="dogbert", golf_swing="fore!")
- ])
-
- c2 = Company(name="Elbonia, Inc.", employees=[
- Engineer(name="vlad", primary_language="cobol")
- ])
-
+ c1 = Company(name='MegaCorp, Inc.',
+ employees=[Engineer(name='dilbert',
+ primary_language='java'), Engineer(name='wally',
+ primary_language='c++'), Manager(name='dogbert',
+ golf_swing='fore!')])
+ c2 = Company(name='Elbonia, Inc.',
+ employees=[Engineer(name='vlad',
+ primary_language='cobol')])
sess.add(c1)
sess.add(c2)
sess.flush()
sess.expunge_all()
+ eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
- eq_((sess.query(Company).
- filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language == 'cobol')).first()),
- c2)
+ # ensure that the Manager mapper was compiled with the Person id
+ # column as higher priority. this ensures that "id" will get
+ # loaded from the Person row and not the possibly non-present
+ # Manager row
+
+ assert Manager.id.property.columns == [Person.__table__.c.id,
+ Manager.__table__.c.id]
+
+ # assert that the "id" column is available without a second
+ # load. this would be the symptom of the previous step not being
+ # correct.
- # ensure that the Manager mapper was compiled
- # with the Person id column as higher priority.
- # this ensures that "id" will get loaded from the Person row
- # and not the possibly non-present Manager row
- assert Manager.id.property.columns == [Person.__table__.c.id, Manager.__table__.c.id]
-
- # assert that the "id" column is available without a second load.
- # this would be the symptom of the previous step not being correct.
sess.expunge_all()
+
def go():
- assert sess.query(Manager).filter(Manager.name=='dogbert').one().id
+ assert sess.query(Manager).filter(Manager.name == 'dogbert'
+ ).one().id
+
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
+
def go():
- assert sess.query(Person).filter(Manager.name=='dogbert').one().id
+ assert sess.query(Person).filter(Manager.name == 'dogbert'
+ ).one().id
+
self.assert_sql_count(testing.db, go, 1)
-
+
def test_add_subcol_after_the_fact(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column('name', String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
-
- Engineer.primary_language = Column('primary_language', String(50))
-
- Base.metadata.create_all()
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
+ Engineer.primary_language = Column('primary_language',
+ String(50))
+ Base.metadata.create_all()
sess = create_session()
e1 = Engineer(primary_language='java', name='dilbert')
sess.add(e1)
sess.flush()
sess.expunge_all()
+ eq_(sess.query(Person).first(), Engineer(primary_language='java'
+ , name='dilbert'))
- eq_(sess.query(Person).first(),
- Engineer(primary_language='java', name='dilbert')
- )
-
def test_add_parentcol_after_the_fact(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
-
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
+
Person.name = Column('name', String(50))
-
Base.metadata.create_all()
-
sess = create_session()
e1 = Engineer(primary_language='java', name='dilbert')
sess.add(e1)
sess.flush()
sess.expunge_all()
-
- eq_(sess.query(Person).first(),
- Engineer(primary_language='java', name='dilbert')
- )
+ eq_(sess.query(Person).first(),
+ Engineer(primary_language='java', name='dilbert'))
def test_add_sub_parentcol_after_the_fact(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
-
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
+
class Admin(Engineer):
+
__tablename__ = 'admins'
- __mapper_args__ = {'polymorphic_identity':'admin'}
+ __mapper_args__ = {'polymorphic_identity': 'admin'}
workstation = Column(String(50))
- id = Column('id', Integer, ForeignKey('engineers.id'), primary_key=True)
-
- Person.name = Column('name', String(50))
+ id = Column('id', Integer, ForeignKey('engineers.id'),
+ primary_key=True)
+ Person.name = Column('name', String(50))
Base.metadata.create_all()
-
sess = create_session()
- e1 = Admin(primary_language='java', name='dilbert', workstation='foo')
+ e1 = Admin(primary_language='java', name='dilbert',
+ workstation='foo')
sess.add(e1)
sess.flush()
sess.expunge_all()
+ eq_(sess.query(Person).first(), Admin(primary_language='java',
+ name='dilbert', workstation='foo'))
- eq_(sess.query(Person).first(),
- Admin(primary_language='java', name='dilbert', workstation='foo')
- )
-
def test_subclass_mixin(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
-
+ __mapper_args__ = {'polymorphic_on': discriminator}
+
class MyMixin(object):
+
pass
-
+
class Engineer(MyMixin, Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ id = Column('id', Integer, ForeignKey('people.id'),
+ primary_key=True)
primary_language = Column('primary_language', String(50))
-
+
assert class_mapper(Engineer).inherits is class_mapper(Person)
-
+
def test_with_undefined_foreignkey(self):
+
class Parent(Base):
- __tablename__ = 'parent'
- id = Column('id', Integer, primary_key=True)
- tp = Column('type', String(50))
- __mapper_args__ = dict(polymorphic_on = tp)
+
+ __tablename__ = 'parent'
+ id = Column('id', Integer, primary_key=True)
+ tp = Column('type', String(50))
+ __mapper_args__ = dict(polymorphic_on=tp)
class Child1(Parent):
- __tablename__ = 'child1'
- id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
- related_child2 = Column('c2', Integer, ForeignKey('child2.id'))
- __mapper_args__ = dict(polymorphic_identity = 'child1')
- # no exception is raised by the ForeignKey to "child2" even though
- # child2 doesn't exist yet
+ __tablename__ = 'child1'
+ id = Column('id', Integer, ForeignKey('parent.id'),
+ primary_key=True)
+ related_child2 = Column('c2', Integer,
+ ForeignKey('child2.id'))
+ __mapper_args__ = dict(polymorphic_identity='child1')
+
+ # no exception is raised by the ForeignKey to "child2" even
+ # though child2 doesn't exist yet
class Child2(Parent):
- __tablename__ = 'child2'
- id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
- related_child1 = Column('c1', Integer)
- __mapper_args__ = dict(polymorphic_identity = 'child2')
+
+ __tablename__ = 'child2'
+ id = Column('id', Integer, ForeignKey('parent.id'),
+ primary_key=True)
+ related_child1 = Column('c1', Integer)
+ __mapper_args__ = dict(polymorphic_identity='child2')
sa.orm.compile_mappers() # no exceptions here
def test_single_colsonbase(self):
- """test single inheritance where all the columns are on the base class."""
-
+ """test single inheritance where all the columns are on the base
+ class."""
+
class Company(Base, ComparableEntity):
+
__tablename__ = 'companies'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column('name', String(50))
- employees = relationship("Person")
+ employees = relationship('Person')
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
company_id = Column('company_id', Integer,
ForeignKey('companies.id'))
name = Column('name', String(50))
discriminator = Column('type', String(50))
primary_language = Column('primary_language', String(50))
golf_swing = Column('golf_swing', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
class Manager(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
- Base.metadata.create_all()
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
+ Base.metadata.create_all()
sess = create_session()
- c1 = Company(name="MegaCorp, Inc.", employees=[
- Engineer(name="dilbert", primary_language="java"),
- Engineer(name="wally", primary_language="c++"),
- Manager(name="dogbert", golf_swing="fore!")
- ])
-
- c2 = Company(name="Elbonia, Inc.", employees=[
- Engineer(name="vlad", primary_language="cobol")
- ])
-
+ c1 = Company(name='MegaCorp, Inc.',
+ employees=[Engineer(name='dilbert',
+ primary_language='java'), Engineer(name='wally',
+ primary_language='c++'), Manager(name='dogbert',
+ golf_swing='fore!')])
+ c2 = Company(name='Elbonia, Inc.',
+ employees=[Engineer(name='vlad',
+ primary_language='cobol')])
sess.add(c1)
sess.add(c2)
sess.flush()
sess.expunge_all()
-
- eq_((sess.query(Person).
- filter(Engineer.primary_language == 'cobol').first()),
- Engineer(name='vlad'))
- eq_((sess.query(Company).
- filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language == 'cobol')).first()),
- c2)
+ eq_(sess.query(Person).filter(Engineer.primary_language
+ == 'cobol').first(), Engineer(name='vlad'))
+ eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
def test_single_colsonsub(self):
- """test single inheritance where the columns are local to their class.
+ """test single inheritance where the columns are local to their
+ class.
this is a newer usage.
"""
class Company(Base, ComparableEntity):
+
__tablename__ = 'companies'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column('name', String(50))
- employees = relationship("Person")
+ employees = relationship('Person')
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
- company_id = Column(Integer,
- ForeignKey('companies.id'))
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ company_id = Column(Integer, ForeignKey('companies.id'))
name = Column(String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
class Manager(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
+
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
golf_swing = Column(String(50))
- # we have here a situation that is somewhat unique.
- # the Person class is mapped to the "people" table, but it
- # was mapped when the table did not include the "primary_language"
- # or "golf_swing" columns. declarative will also manipulate
- # the exclude_properties collection so that sibling classes
- # don't cross-pollinate.
+ # we have here a situation that is somewhat unique. the Person
+ # class is mapped to the "people" table, but it was mapped when
+ # the table did not include the "primary_language" or
+ # "golf_swing" columns. declarative will also manipulate the
+ # exclude_properties collection so that sibling classes don't
+ # cross-pollinate.
assert Person.__table__.c.company_id is not None
assert Person.__table__.c.golf_swing is not None
assert not hasattr(Person, 'golf_swing')
assert not hasattr(Engineer, 'golf_swing')
assert not hasattr(Manager, 'primary_language')
-
Base.metadata.create_all()
-
sess = create_session()
-
- e1 = Engineer(name="dilbert", primary_language="java")
- e2 = Engineer(name="wally", primary_language="c++")
- m1 = Manager(name="dogbert", golf_swing="fore!")
- c1 = Company(name="MegaCorp, Inc.", employees=[e1, e2, m1])
-
- e3 =Engineer(name="vlad", primary_language="cobol")
- c2 = Company(name="Elbonia, Inc.", employees=[e3])
+ e1 = Engineer(name='dilbert', primary_language='java')
+ e2 = Engineer(name='wally', primary_language='c++')
+ m1 = Manager(name='dogbert', golf_swing='fore!')
+ c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1])
+ e3 = Engineer(name='vlad', primary_language='cobol')
+ c2 = Company(name='Elbonia, Inc.', employees=[e3])
sess.add(c1)
sess.add(c2)
sess.flush()
sess.expunge_all()
-
- eq_((sess.query(Person).
- filter(Engineer.primary_language == 'cobol').first()),
- Engineer(name='vlad'))
- eq_((sess.query(Company).
- filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language == 'cobol')).first()),
- c2)
-
- eq_(
- sess.query(Engineer).filter_by(primary_language='cobol').one(),
- Engineer(name="vlad", primary_language="cobol")
- )
+ eq_(sess.query(Person).filter(Engineer.primary_language
+ == 'cobol').first(), Engineer(name='vlad'))
+ eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
+ eq_(sess.query(Engineer).filter_by(primary_language='cobol'
+ ).one(), Engineer(name='vlad', primary_language='cobol'))
def test_joined_from_single(self):
+
class Company(Base, ComparableEntity):
+
__tablename__ = 'companies'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column('name', String(50))
- employees = relationship("Person")
-
+ employees = relationship('Person')
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
company_id = Column(Integer, ForeignKey('companies.id'))
name = Column(String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Manager(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
+
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
golf_swing = Column(String(50))
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- id = Column(Integer, ForeignKey('people.id'), primary_key=True)
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ id = Column(Integer, ForeignKey('people.id'),
+ primary_key=True)
primary_language = Column(String(50))
assert Person.__table__.c.golf_swing is not None
assert not hasattr(Person, 'golf_swing')
assert not hasattr(Engineer, 'golf_swing')
assert not hasattr(Manager, 'primary_language')
-
Base.metadata.create_all()
-
sess = create_session()
-
- e1 = Engineer(name="dilbert", primary_language="java")
- e2 = Engineer(name="wally", primary_language="c++")
- m1 = Manager(name="dogbert", golf_swing="fore!")
- c1 = Company(name="MegaCorp, Inc.", employees=[e1, e2, m1])
- e3 =Engineer(name="vlad", primary_language="cobol")
- c2 = Company(name="Elbonia, Inc.", employees=[e3])
+ e1 = Engineer(name='dilbert', primary_language='java')
+ e2 = Engineer(name='wally', primary_language='c++')
+ m1 = Manager(name='dogbert', golf_swing='fore!')
+ c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1])
+ e3 = Engineer(name='vlad', primary_language='cobol')
+ c2 = Company(name='Elbonia, Inc.', employees=[e3])
sess.add(c1)
sess.add(c2)
sess.flush()
sess.expunge_all()
-
- eq_((sess.query(Person).with_polymorphic(Engineer).
- filter(Engineer.primary_language == 'cobol').first()),
- Engineer(name='vlad'))
- eq_((sess.query(Company).
- filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language == 'cobol')).first()),
- c2)
-
- eq_(
- sess.query(Engineer).filter_by(primary_language='cobol').one(),
- Engineer(name="vlad", primary_language="cobol")
- )
+ eq_(sess.query(Person).with_polymorphic(Engineer).
+ filter(Engineer.primary_language
+ == 'cobol').first(), Engineer(name='vlad'))
+ eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
+ eq_(sess.query(Engineer).filter_by(primary_language='cobol'
+ ).one(), Engineer(name='vlad', primary_language='cobol'))
def test_add_deferred(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
Person.name = deferred(Column(String(10)))
-
Base.metadata.create_all()
sess = create_session()
p = Person(name='ratbert')
-
sess.add(p)
sess.flush()
sess.expunge_all()
- eq_(
- sess.query(Person).all(),
- [
- Person(name='ratbert')
- ]
- )
+ eq_(sess.query(Person).all(), [Person(name='ratbert')])
sess.expunge_all()
-
- person = sess.query(Person).filter(Person.name == 'ratbert').one()
+ person = sess.query(Person).filter(Person.name == 'ratbert'
+ ).one()
assert 'name' not in person.__dict__
def test_single_fksonsub(self):
- """test single inheritance with a foreign key-holding column on a subclass.
-
+ """test single inheritance with a foreign key-holding column on
+ a subclass.
+
"""
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column(String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- primary_language_id = Column(Integer, ForeignKey('languages.id'))
- primary_language = relationship("Language")
-
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ primary_language_id = Column(Integer,
+ ForeignKey('languages.id'))
+ primary_language = relationship('Language')
+
class Language(Base, ComparableEntity):
+
__tablename__ = 'languages'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column(String(50))
assert not hasattr(Person, 'primary_language_id')
-
Base.metadata.create_all()
-
sess = create_session()
-
- java, cpp, cobol = Language(name='java'),Language(name='cpp'), Language(name='cobol')
- e1 = Engineer(name="dilbert", primary_language=java)
- e2 = Engineer(name="wally", primary_language=cpp)
- e3 =Engineer(name="vlad", primary_language=cobol)
+ java, cpp, cobol = Language(name='java'), Language(name='cpp'), \
+ Language(name='cobol')
+ e1 = Engineer(name='dilbert', primary_language=java)
+ e2 = Engineer(name='wally', primary_language=cpp)
+ e3 = Engineer(name='vlad', primary_language=cobol)
sess.add_all([e1, e2, e3])
sess.flush()
sess.expunge_all()
+ eq_(sess.query(Person).filter(Engineer.primary_language.has(
+ Language.name
+ == 'cobol')).first(), Engineer(name='vlad',
+ primary_language=Language(name='cobol')))
+ eq_(sess.query(Engineer).filter(Engineer.primary_language.has(
+ Language.name
+ == 'cobol')).one(), Engineer(name='vlad',
+ primary_language=Language(name='cobol')))
+ eq_(sess.query(Person).join(Engineer.primary_language).order_by(
+ Language.name).all(),
+ [Engineer(name='vlad',
+ primary_language=Language(name='cobol')),
+ Engineer(name='wally', primary_language=Language(name='cpp'
+ )), Engineer(name='dilbert',
+ primary_language=Language(name='java'))])
- eq_((sess.query(Person).
- filter(Engineer.primary_language.has(Language.name=='cobol')).first()),
- Engineer(name='vlad', primary_language=Language(name='cobol')))
-
- eq_(
- sess.query(Engineer).filter(Engineer.primary_language.has(Language.name=='cobol')).one(),
- Engineer(name="vlad", primary_language=Language(name='cobol'))
- )
-
- eq_(
- sess.query(Person).join(Engineer.primary_language).order_by(Language.name).all(),
- [
- Engineer(name='vlad', primary_language=Language(name='cobol')),
- Engineer(name='wally', primary_language=Language(name='cpp')),
- Engineer(name='dilbert', primary_language=Language(name='java')),
- ]
- )
-
def test_single_three_levels(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
name = Column(String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
class JuniorEngineer(Engineer):
- __mapper_args__ = {'polymorphic_identity':'junior_engineer'}
+
+ __mapper_args__ = \
+ {'polymorphic_identity': 'junior_engineer'}
nerf_gun = Column(String(50))
class Manager(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
+
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
golf_swing = Column(String(50))
assert JuniorEngineer.nerf_gun
assert not hasattr(Engineer, 'nerf_gun')
assert not hasattr(Manager, 'nerf_gun')
assert not hasattr(Manager, 'primary_language')
-
+
def test_single_detects_conflict(self):
+
class Person(Base):
+
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
name = Column(String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
-
+
# test sibling col conflict
+
def go():
+
class Manager(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
+
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
golf_swing = Column(String(50))
primary_language = Column(String(50))
+
assert_raises(sa.exc.ArgumentError, go)
# test parent col conflict
+
def go():
+
class Salesman(Person):
- __mapper_args__ = {'polymorphic_identity':'manager'}
+
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
name = Column(String(50))
+
assert_raises(sa.exc.ArgumentError, go)
-
def test_single_no_special_cols(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
def go():
+
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- primary_language = Column('primary_language', String(50))
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ primary_language = Column('primary_language',
+ String(50))
foo_bar = Column(Integer, primary_key=True)
- assert_raises_message(sa.exc.ArgumentError, "place primary key", go)
-
+
+ assert_raises_message(sa.exc.ArgumentError, 'place primary key'
+ , go)
+
def test_single_no_table_args(self):
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
discriminator = Column('type', String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
def go():
+
class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity':'engineer'}
- primary_language = Column('primary_language', String(50))
+
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+ primary_language = Column('primary_language',
+ String(50))
+
# this should be on the Person class, as this is single
# table inheritance, which is why we test that this
# throws an exception!
- __table_args__ = {'mysql_engine':'InnoDB'}
- assert_raises_message(sa.exc.ArgumentError, "place __table_args__", go)
-
+
+ __table_args__ = {'mysql_engine': 'InnoDB'}
+
+ assert_raises_message(sa.exc.ArgumentError,
+ 'place __table_args__', go)
+
def test_concrete(self):
- engineers = Table('engineers', Base.metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(50)),
- Column('primary_language', String(50))
- )
- managers = Table('managers', Base.metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(50)),
- Column('golf_swing', String(50))
- )
-
- punion = polymorphic_union({
- 'engineer':engineers,
- 'manager':managers
- }, 'type', 'punion')
+ engineers = Table('engineers', Base.metadata, Column('id',
+ Integer, primary_key=True,
+ test_needs_autoincrement=True), Column('name'
+ , String(50)), Column('primary_language',
+ String(50)))
+ managers = Table('managers', Base.metadata, Column('id',
+ Integer, primary_key=True,
+ test_needs_autoincrement=True), Column('name',
+ String(50)), Column('golf_swing', String(50)))
+ punion = polymorphic_union({'engineer': engineers, 'manager'
+ : managers}, 'type', 'punion')
class Person(Base, ComparableEntity):
+
__table__ = punion
- __mapper_args__ = {'polymorphic_on':punion.c.type}
+ __mapper_args__ = {'polymorphic_on': punion.c.type}
class Engineer(Person):
+
__table__ = engineers
- __mapper_args__ = {'polymorphic_identity':'engineer', 'concrete':True}
+ __mapper_args__ = {'polymorphic_identity': 'engineer',
+ 'concrete': True}
class Manager(Person):
+
__table__ = managers
- __mapper_args__ = {'polymorphic_identity':'manager', 'concrete':True}
-
+ __mapper_args__ = {'polymorphic_identity': 'manager',
+ 'concrete': True}
+
Base.metadata.create_all()
sess = create_session()
-
- e1 = Engineer(name="dilbert", primary_language="java")
- e2 = Engineer(name="wally", primary_language="c++")
- m1 = Manager(name="dogbert", golf_swing="fore!")
- e3 = Engineer(name="vlad", primary_language="cobol")
-
+ e1 = Engineer(name='dilbert', primary_language='java')
+ e2 = Engineer(name='wally', primary_language='c++')
+ m1 = Manager(name='dogbert', golf_swing='fore!')
+ e3 = Engineer(name='vlad', primary_language='cobol')
sess.add_all([e1, e2, m1, e3])
sess.flush()
sess.expunge_all()
- eq_(
- sess.query(Person).order_by(Person.name).all(),
- [
- Engineer(name='dilbert'), Manager(name='dogbert'),
- Engineer(name='vlad'), Engineer(name='wally')
- ]
- )
-
+ eq_(sess.query(Person).order_by(Person.name).all(),
+ [Engineer(name='dilbert'), Manager(name='dogbert'),
+ Engineer(name='vlad'), Engineer(name='wally')])
+
def test_concrete_inline_non_polymorphic(self):
"""test the example from the declarative docs."""
-
+
class Person(Base, ComparableEntity):
+
__tablename__ = 'people'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
name = Column(String(50))
class Engineer(Person):
+
__tablename__ = 'engineers'
- __mapper_args__ = {'concrete':True}
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ __mapper_args__ = {'concrete': True}
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
primary_language = Column(String(50))
name = Column(String(50))
class Manager(Person):
+
__tablename__ = 'manager'
- __mapper_args__ = {'concrete':True}
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ __mapper_args__ = {'concrete': True}
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
golf_swing = Column(String(50))
name = Column(String(50))
Base.metadata.create_all()
sess = create_session()
-
- e1 = Engineer(name="dilbert", primary_language="java")
- e2 = Engineer(name="wally", primary_language="c++")
- m1 = Manager(name="dogbert", golf_swing="fore!")
- e3 = Engineer(name="vlad", primary_language="cobol")
-
+ e1 = Engineer(name='dilbert', primary_language='java')
+ e2 = Engineer(name='wally', primary_language='c++')
+ m1 = Manager(name='dogbert', golf_swing='fore!')
+ e3 = Engineer(name='vlad', primary_language='cobol')
sess.add_all([e1, e2, m1, e3])
sess.flush()
sess.expunge_all()
- eq_(
- sess.query(Engineer).order_by(Engineer.name).all(),
- [
- Engineer(name='dilbert'),
- Engineer(name='vlad'), Engineer(name='wally')
- ]
- )
- eq_(
- sess.query(Manager).all(),
- [
- Manager(name='dogbert'),
- ]
- )
-
-
+ eq_(sess.query(Engineer).order_by(Engineer.name).all(),
+ [Engineer(name='dilbert'), Engineer(name='vlad'),
+ Engineer(name='wally')])
+ eq_(sess.query(Manager).all(), [Manager(name='dogbert')])
+
def _produce_test(inline, stringbased):
+
class ExplicitJoinTest(MappedTest):
-
+
@classmethod
def define_tables(cls, metadata):
global User, Address
Base = decl.declarative_base(metadata=metadata)
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
-
+
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
if inline:
if stringbased:
- user = relationship("User",
- primaryjoin="User.id==Address.user_id",
- backref="addresses")
+ user = relationship('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:
compile_mappers()
if stringbased:
- Address.user = relationship("User",
- primaryjoin="User.id==Address.user_id",
- backref="addresses")
+ Address.user = relationship('User',
+ primaryjoin='User.id==Address.user_id',
+ backref='addresses')
else:
- Address.user = relationship(User,
- primaryjoin=User.id==Address.user_id,
- backref="addresses")
+ Address.user = relationship(User,
+ primaryjoin=User.id == Address.user_id,
+ backref='addresses')
@classmethod
def insert_data(cls):
- params = [dict(zip(('id', 'name'), column_values))
- for column_values in
- [(7, 'jack'),
- (8, 'ed'),
- (9, 'fred'),
- (10, 'chuck')]
- ]
+ params = [dict(zip(('id', 'name'), column_values))
+ for column_values in [(7, 'jack'), (8, 'ed'), (9,
+ 'fred'), (10, 'chuck')]]
User.__table__.insert().execute(params)
-
- Address.__table__.insert().execute(
- [dict(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(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):
- # this query will screw up if the aliasing
- # enabled in query.join() gets applied to the right half of the
- # join condition inside the any().
- # the join condition inside of any() comes from the
- # "primaryjoin" of the relationship,
- # and should not be annotated with _orm_adapt.
- # PropertyLoader.Comparator will annotate
- # the left side with _orm_adapt, though.
+
+ # this query will screw up if the aliasing enabled in
+ # query.join() gets applied to the right half of the join
+ # condition inside the any(). the join condition inside of
+ # any() comes from the "primaryjoin" of the relationship,
+ # and should not be annotated with _orm_adapt.
+ # PropertyLoader.Comparator will annotate the left side with
+ # _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):
+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
-
+
class DeclarativeReflectionTest(testing.TestBase):
+
@classmethod
def setup_class(cls):
global reflection_metadata
reflection_metadata = MetaData(testing.db)
-
- Table('users', reflection_metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(50)),
- test_needs_fk=True)
- Table('addresses', reflection_metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('email', String(50)),
- Column('user_id', Integer, ForeignKey('users.id')),
- test_needs_fk=True)
- Table('imhandles', reflection_metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('user_id', Integer),
- Column('network', String(50)),
- Column('handle', String(50)),
- test_needs_fk=True)
-
+ Table('users', reflection_metadata, Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)), test_needs_fk=True)
+ Table(
+ 'addresses',
+ reflection_metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('email', String(50)),
+ Column('user_id', Integer, ForeignKey('users.id')),
+ test_needs_fk=True,
+ )
+ Table(
+ 'imhandles',
+ reflection_metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('user_id', Integer),
+ Column('network', String(50)),
+ Column('handle', String(50)),
+ test_needs_fk=True,
+ )
reflection_metadata.create_all()
def setup(self):
meta = MetaData(testing.db)
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
__autoload__ = True
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
- addresses = relationship("Address", backref="user")
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ addresses = relationship('Address', backref='user')
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
__autoload__ = True
-
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
-
- u1 = User(name='u1', addresses=[
- Address(email='one'),
- Address(email='two'),
- ])
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
+ 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'))
meta = MetaData(testing.db)
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
__autoload__ = True
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
nom = Column('name', String(50), key='nom')
- addresses = relationship("Address", backref="user")
+ addresses = relationship('Address', backref='user')
class Address(Base, ComparableEntity):
+
__tablename__ = 'addresses'
__autoload__ = True
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
- u1 = User(nom='u1', addresses=[
- Address(email='one'),
- Address(email='two'),
- ])
+ u1 = User(nom='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(nom='u1', addresses=[
- Address(email='one'),
- Address(email='two'),
- ])])
-
+ eq_(sess.query(User).all(), [User(nom='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(nom='u1'))
-
assert_raises(TypeError, User, name='u3')
def test_supplied_fk(self):
meta = MetaData(testing.db)
class IMHandle(Base, ComparableEntity):
+
__tablename__ = 'imhandles'
__autoload__ = True
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ user_id = Column('user_id', Integer, ForeignKey('users.id'))
- user_id = Column('user_id', Integer,
- ForeignKey('users.id'))
class User(Base, ComparableEntity):
+
__tablename__ = 'users'
__autoload__ = True
if testing.against('oracle', 'firebird'):
- id = Column('id', Integer, primary_key=True, test_needs_autoincrement=True)
- handles = relationship("IMHandle", backref="user")
+ id = Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ handles = relationship('IMHandle', backref='user')
- u1 = User(name='u1', handles=[
- IMHandle(network='blabber', handle='foo'),
- IMHandle(network='lol', handle='zomg')
- ])
+ u1 = User(name='u1', handles=[IMHandle(network='blabber',
+ handle='foo'), IMHandle(network='lol', handle='zomg'
+ )])
sess = create_session()
sess.add(u1)
sess.flush()
sess.expunge_all()
-
- eq_(sess.query(User).all(), [User(name='u1', handles=[
- IMHandle(network='blabber', handle='foo'),
- IMHandle(network='lol', handle='zomg')
- ])])
-
- a1 = sess.query(IMHandle).filter(IMHandle.handle == 'zomg').one()
+ eq_(sess.query(User).all(), [User(name='u1',
+ handles=[IMHandle(network='blabber', handle='foo'),
+ IMHandle(network='lol', handle='zomg')])])
+ a1 = sess.query(IMHandle).filter(IMHandle.handle == 'zomg'
+ ).one()
eq_(a1, IMHandle(network='lol', handle='zomg'))
eq_(a1.user, User(name='u1'))
class DeclarativeMixinTest(DeclarativeTestBase):
-
+
def test_simple(self):
class MyMixin(object):
- id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
def foo(self):
- return 'bar'+str(self.id)
+ return 'bar' + str(self.id)
+
+ class MyModel(Base, MyMixin):
- class MyModel(Base,MyMixin):
- __tablename__='test'
+ __tablename__ = 'test'
name = Column(String(100), nullable=False, index=True)
Base.metadata.create_all()
-
session = create_session()
session.add(MyModel(name='testing'))
session.flush()
session.expunge_all()
-
obj = session.query(MyModel).one()
- eq_(obj.id,1)
- eq_(obj.name,'testing')
- eq_(obj.foo(),'bar1')
+ eq_(obj.id, 1)
+ eq_(obj.name, 'testing')
+ eq_(obj.foo(), 'bar1')
-
def test_unique_column(self):
-
+
class MyMixin(object):
+
id = Column(Integer, primary_key=True)
value = Column(String, unique=True)
class MyModel(Base, MyMixin):
+
__tablename__ = 'test'
-
+
assert MyModel.__table__.c.value.unique
-
+
def test_hierarchical_bases(self):
class MyMixinParent:
- id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
def foo(self):
- return 'bar'+str(self.id)
+ return 'bar' + str(self.id)
class MyMixin(MyMixinParent):
+
baz = Column(String(100), nullable=False, index=True)
- class MyModel(Base,MyMixin):
- __tablename__='test'
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
name = Column(String(100), nullable=False, index=True)
Base.metadata.create_all()
-
session = create_session()
session.add(MyModel(name='testing', baz='fu'))
session.flush()
session.expunge_all()
-
obj = session.query(MyModel).one()
- eq_(obj.id,1)
- eq_(obj.name,'testing')
- eq_(obj.foo(),'bar1')
- eq_(obj.baz,'fu')
-
+ eq_(obj.id, 1)
+ eq_(obj.name, 'testing')
+ eq_(obj.foo(), 'bar1')
+ eq_(obj.baz, 'fu')
+
def test_not_allowed(self):
+
class MyMixin:
foo = Column(Integer, ForeignKey('bar.id'))
-
+
def go():
class MyModel(Base, MyMixin):
__tablename__ = 'foo'
-
+
assert_raises(sa.exc.InvalidRequestError, go)
-
+
class MyRelMixin:
- foo = relationship("Bar")
+ foo = relationship('Bar')
+
def go():
class MyModel(Base, MyRelMixin):
+
__tablename__ = 'foo'
+
assert_raises(sa.exc.InvalidRequestError, go)
-
+
class MyDefMixin:
foo = deferred(Column('foo', String))
+
def go():
class MyModel(Base, MyDefMixin):
__tablename__ = 'foo'
+
assert_raises(sa.exc.InvalidRequestError, go)
class MyCPropMixin:
foo = column_property(Column('foo', String))
+
def go():
class MyModel(Base, MyCPropMixin):
__tablename__ = 'foo'
+
assert_raises(sa.exc.InvalidRequestError, go)
-
+
def test_table_name_inherited(self):
-
+
class MyMixin:
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
- id = Column(Integer, primary_key=True)
+ id = Column(Integer, primary_key=True)
- class MyModel(Base,MyMixin):
+ class MyModel(Base, MyMixin):
pass
- eq_(MyModel.__table__.name,'mymodel')
-
+ eq_(MyModel.__table__.name, 'mymodel')
+
def test_table_name_not_inherited(self):
-
+
class MyMixin:
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
- id = Column(Integer, primary_key=True)
+ id = Column(Integer, primary_key=True)
- class MyModel(Base,MyMixin):
+ class MyModel(Base, MyMixin):
__tablename__ = 'overridden'
- eq_(MyModel.__table__.name,'overridden')
-
+ eq_(MyModel.__table__.name, 'overridden')
+
def test_table_name_inheritance_order(self):
-
+
class MyMixin1:
@classproperty
def __tablename__(cls):
- return cls.__name__.lower()+'1'
+ return cls.__name__.lower() + '1'
class MyMixin2:
@classproperty
def __tablename__(cls):
- return cls.__name__.lower()+'2'
-
- class MyModel(Base,MyMixin1,MyMixin2):
- id = Column(Integer, primary_key=True)
+ return cls.__name__.lower() + '2'
+
+ class MyModel(Base, MyMixin1, MyMixin2):
+ id = Column(Integer, primary_key=True)
+
+ eq_(MyModel.__table__.name, 'mymodel1')
- eq_(MyModel.__table__.name,'mymodel1')
-
def test_table_name_dependent_on_subclass(self):
+
class MyHistoryMixin:
@classproperty
def __tablename__(cls):
class MyModel(Base, MyHistoryMixin):
parent_name = 'foo'
id = Column(Integer, primary_key=True)
-
+
eq_(MyModel.__table__.name, 'foo_changelog')
-
+
def test_table_args_inherited(self):
-
+
class MyMixin:
- __table_args__ = {'mysql_engine':'InnoDB'}
+ __table_args__ = {'mysql_engine': 'InnoDB'}
+
+ class MyModel(Base, MyMixin):
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ eq_(MyModel.__table__.kwargs, {'mysql_engine': 'InnoDB'})
- eq_(MyModel.__table__.kwargs,{'mysql_engine': 'InnoDB'})
-
def test_table_args_inherited_descriptor(self):
-
+
class MyMixin:
@classproperty
def __table_args__(cls):
- return {'info':cls.__name__}
+ return {'info': cls.__name__}
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ class MyModel(Base, MyMixin):
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
+
+ eq_(MyModel.__table__.info, 'MyModel')
- eq_(MyModel.__table__.info,'MyModel')
-
def test_table_args_inherited_single_table_inheritance(self):
-
+
class MyMixin:
- __table_args__ = {'mysql_engine':'InnoDB'}
+ __table_args__ = {'mysql_engine': 'InnoDB'}
- class General(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ class General(Base, MyMixin):
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
type_ = Column(String(50))
- __mapper__args = {'polymorphic_on':type_}
+ __mapper__args = {'polymorphic_on': type_}
class Specific(General):
- __mapper_args__ = {'polymorphic_identity':'specific'}
+ __mapper_args__ = {'polymorphic_identity': 'specific'}
assert Specific.__table__ is General.__table__
- eq_(General.__table__.kwargs,{'mysql_engine': 'InnoDB'})
-
+ eq_(General.__table__.kwargs, {'mysql_engine': 'InnoDB'})
+
def test_table_args_overridden(self):
-
+
class MyMixin:
- __table_args__ = {'mysql_engine':'Foo'}
+ __table_args__ = {'mysql_engine': 'Foo'}
+
+ class MyModel(Base, MyMixin):
+ __tablename__ = 'test'
+ __table_args__ = {'mysql_engine': 'InnoDB'}
+ id = Column(Integer, primary_key=True)
- class MyModel(Base,MyMixin):
- __tablename__='test'
- __table_args__ = {'mysql_engine':'InnoDB'}
- id = Column(Integer, primary_key=True)
+ eq_(MyModel.__table__.kwargs, {'mysql_engine': 'InnoDB'})
- eq_(MyModel.__table__.kwargs,{'mysql_engine': 'InnoDB'})
-
def test_mapper_args_classproperty(self):
+
class ComputedMapperArgs:
@classproperty
def __mapper_args__(cls):
- if cls.__name__=='Person':
- return {'polymorphic_on':cls.discriminator}
+ if cls.__name__ == 'Person':
+ return {'polymorphic_on': cls.discriminator}
else:
- return {'polymorphic_identity':cls.__name__}
+ return {'polymorphic_identity': cls.__name__}
- class Person(Base,ComputedMapperArgs):
+ class Person(Base, ComputedMapperArgs):
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
discriminator = Column('type', String(50))
pass
compile_mappers()
-
- assert class_mapper(Person).polymorphic_on is Person.__table__.c.type
+ assert class_mapper(Person).polymorphic_on \
+ is Person.__table__.c.type
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
def test_mapper_args_classproperty_two(self):
- # same as test_mapper_args_classproperty, but
- # we repeat ComputedMapperArgs on both classes
- # for no apparent reason.
-
+
+ # same as test_mapper_args_classproperty, but we repeat
+ # ComputedMapperArgs on both classes for no apparent reason.
+
class ComputedMapperArgs:
@classproperty
def __mapper_args__(cls):
- if cls.__name__=='Person':
- return {'polymorphic_on':cls.discriminator}
+ if cls.__name__ == 'Person':
+ return {'polymorphic_on': cls.discriminator}
else:
- return {'polymorphic_identity':cls.__name__}
+ return {'polymorphic_identity': cls.__name__}
+
+ class Person(Base, ComputedMapperArgs):
- class Person(Base,ComputedMapperArgs):
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
discriminator = Column('type', String(50))
pass
compile_mappers()
-
- assert class_mapper(Person).polymorphic_on is Person.__table__.c.type
+ assert class_mapper(Person).polymorphic_on \
+ is Person.__table__.c.type
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
-
+
def test_table_args_composite(self):
class MyMixin1:
- __table_args__ = {'info':{'baz':'bob'}}
+
+ __table_args__ = {'info': {'baz': 'bob'}}
class MyMixin2:
- __table_args__ = {'info':{'foo':'bar'}}
- class MyModel(Base,MyMixin1,MyMixin2):
- __tablename__='test'
+ __table_args__ = {'info': {'foo': 'bar'}}
+
+ class MyModel(Base, MyMixin1, MyMixin2):
+
+ __tablename__ = 'test'
@classproperty
def __table_args__(self):
info.update(MyMixin1.__table_args__['info'])
info.update(MyMixin2.__table_args__['info'])
return args
-
- id = Column(Integer, primary_key=True)
+ id = Column(Integer, primary_key=True)
+
+ eq_(MyModel.__table__.info, {'foo': 'bar', 'baz': 'bob'})
- eq_(MyModel.__table__.info,{
- 'foo': 'bar',
- 'baz': 'bob',
- })
-
def test_mapper_args_inherited(self):
-
+
class MyMixin:
- __mapper_args__ = {'always_refresh':True}
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ __mapper_args__ = {'always_refresh': True}
+
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
+
+ eq_(MyModel.__mapper__.always_refresh, True)
- eq_(MyModel.__mapper__.always_refresh,True)
-
-
def test_mapper_args_inherited_descriptor(self):
-
+
class MyMixin:
+
@classproperty
def __mapper_args__(cls):
+
# tenuous, but illustrates the problem!
- if cls.__name__=='MyModel':
+
+ if cls.__name__ == 'MyModel':
return dict(always_refresh=True)
else:
return dict(always_refresh=False)
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
+
+ eq_(MyModel.__mapper__.always_refresh, True)
- eq_(MyModel.__mapper__.always_refresh,True)
-
def test_mapper_args_polymorphic_on_inherited(self):
class MyMixin:
+
type_ = Column(String(50))
- __mapper_args__= {'polymorphic_on':type_}
+ __mapper_args__ = {'polymorphic_on': type_}
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
col = MyModel.__mapper__.polymorphic_on
- eq_(col.name,'type_')
+ eq_(col.name, 'type_')
assert col.table is not None
def test_mapper_args_overridden(self):
-
+
class MyMixin:
- __mapper_args__=dict(always_refresh=True)
- class MyModel(Base,MyMixin):
- __tablename__='test'
- __mapper_args__=dict(always_refresh=False)
- id = Column(Integer, primary_key=True)
+ __mapper_args__ = dict(always_refresh=True)
+
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ __mapper_args__ = dict(always_refresh=False)
+ id = Column(Integer, primary_key=True)
- eq_(MyModel.__mapper__.always_refresh,False)
+ eq_(MyModel.__mapper__.always_refresh, False)
def test_mapper_args_composite(self):
class MyMixin1:
+
type_ = Column(String(50))
- __mapper_args__ = {'polymorphic_on':type_}
+ __mapper_args__ = {'polymorphic_on': type_}
class MyMixin2:
- __mapper_args__ = {'always_refresh':True}
- class MyModel(Base,MyMixin1,MyMixin2):
- __tablename__='test'
+ __mapper_args__ = {'always_refresh': True}
+
+ class MyModel(Base, MyMixin1, MyMixin2):
+
+ __tablename__ = 'test'
@classproperty
def __mapper_args__(self):
args.update(MyMixin1.__mapper_args__)
args.update(MyMixin2.__mapper_args__)
return args
-
- id = Column(Integer, primary_key=True)
-
+ id = Column(Integer, primary_key=True)
+
col = MyModel.__mapper__.polymorphic_on
- eq_(col.name,'type_')
+ eq_(col.name, 'type_')
assert col.table is not None
-
- eq_(MyModel.__mapper__.always_refresh,True)
+ eq_(MyModel.__mapper__.always_refresh, True)
def test_single_table_no_propagation(self):
class IdColumn:
+
id = Column(Integer, primary_key=True)
class Generic(Base, IdColumn):
+
__tablename__ = 'base'
discriminator = Column('type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
+ __mapper_args__ = dict(polymorphic_on=discriminator)
value = Column(Integer())
class Specific(Generic):
+
__mapper_args__ = dict(polymorphic_identity='specific')
assert Specific.__table__ is Generic.__table__
- eq_(Generic.__table__.c.keys(),['id', 'type', 'value'])
- assert class_mapper(Specific).polymorphic_on is \
- Generic.__table__.c.type
+ eq_(Generic.__table__.c.keys(), ['id', 'type', 'value'])
+ assert class_mapper(Specific).polymorphic_on \
+ is Generic.__table__.c.type
eq_(class_mapper(Specific).polymorphic_identity, 'specific')
def test_joined_table_propagation(self):
class CommonMixin:
-
+
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
-
- __table_args__ = {'mysql_engine':'InnoDB'}
-
- timestamp = Column(Integer)
+ __table_args__ = {'mysql_engine': 'InnoDB'}
+ timestamp = Column(Integer)
id = Column(Integer, primary_key=True)
-
+
class Generic(Base, CommonMixin):
+
discriminator = Column('python_type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
+ __mapper_args__ = dict(polymorphic_on=discriminator)
class Specific(Generic):
+
__mapper_args__ = dict(polymorphic_identity='specific')
- id = Column(Integer, ForeignKey('generic.id'), primary_key=True)
- eq_(Generic.__table__.name,'generic')
- eq_(Specific.__table__.name,'specific')
- eq_(Generic.__table__.c.keys(),['timestamp', 'id', 'python_type'])
- eq_(Specific.__table__.c.keys(),['timestamp', 'id'])
- eq_(Generic.__table__.kwargs,{'mysql_engine': 'InnoDB'})
- eq_(Specific.__table__.kwargs,{'mysql_engine': 'InnoDB'})
-
+ id = Column(Integer, ForeignKey('generic.id'),
+ primary_key=True)
+
+ eq_(Generic.__table__.name, 'generic')
+ eq_(Specific.__table__.name, 'specific')
+ eq_(Generic.__table__.c.keys(), ['timestamp', 'id',
+ 'python_type'])
+ eq_(Specific.__table__.c.keys(), ['timestamp', 'id'])
+ eq_(Generic.__table__.kwargs, {'mysql_engine': 'InnoDB'})
+ eq_(Specific.__table__.kwargs, {'mysql_engine': 'InnoDB'})
+
def test_some_propagation(self):
-
+
class CommonMixin:
+
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
- __table_args__ = {'mysql_engine':'InnoDB'}
- timestamp = Column(Integer)
+ __table_args__ = {'mysql_engine': 'InnoDB'}
+ timestamp = Column(Integer)
class BaseType(Base, CommonMixin):
+
discriminator = Column('type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
- id = Column(Integer, primary_key=True)
- value = Column(Integer())
+ __mapper_args__ = dict(polymorphic_on=discriminator)
+ id = Column(Integer, primary_key=True)
+ value = Column(Integer())
class Single(BaseType):
+
__tablename__ = None
__mapper_args__ = dict(polymorphic_identity='type1')
class Joined(BaseType):
- __mapper_args__ = dict(polymorphic_identity='type2')
- id = Column(Integer, ForeignKey('basetype.id'), primary_key=True)
- eq_(BaseType.__table__.name,'basetype')
- eq_(BaseType.__table__.c.keys(),
- ['timestamp', 'type', 'id', 'value', ])
- eq_(BaseType.__table__.kwargs,{'mysql_engine': 'InnoDB'})
+ __mapper_args__ = dict(polymorphic_identity='type2')
+ id = Column(Integer, ForeignKey('basetype.id'),
+ primary_key=True)
+ eq_(BaseType.__table__.name, 'basetype')
+ eq_(BaseType.__table__.c.keys(), ['timestamp', 'type', 'id',
+ 'value'])
+ eq_(BaseType.__table__.kwargs, {'mysql_engine': 'InnoDB'})
assert Single.__table__ is BaseType.__table__
+ eq_(Joined.__table__.name, 'joined')
+ eq_(Joined.__table__.c.keys(), ['timestamp', 'id'])
+ eq_(Joined.__table__.kwargs, {'mysql_engine': 'InnoDB'})
- eq_(Joined.__table__.name,'joined')
- eq_(Joined.__table__.c.keys(),['timestamp','id'])
- eq_(Joined.__table__.kwargs,{'mysql_engine': 'InnoDB'})
-
def test_non_propagating_mixin(self):
class NoJoinedTableNameMixin:
+
@classproperty
def __tablename__(cls):
if decl.has_inherited_table(cls):
return cls.__name__.lower()
class BaseType(Base, NoJoinedTableNameMixin):
+
discriminator = Column('type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
- id = Column(Integer, primary_key=True)
- value = Column(Integer())
+ __mapper_args__ = dict(polymorphic_on=discriminator)
+ id = Column(Integer, primary_key=True)
+ value = Column(Integer())
class Specific(BaseType):
- __mapper_args__ = dict(polymorphic_identity='specific')
- eq_(BaseType.__table__.name,'basetype')
- eq_(BaseType.__table__.c.keys(),['type', 'id', 'value'])
+ __mapper_args__ = dict(polymorphic_identity='specific')
+ eq_(BaseType.__table__.name, 'basetype')
+ eq_(BaseType.__table__.c.keys(), ['type', 'id', 'value'])
assert Specific.__table__ is BaseType.__table__
- assert class_mapper(Specific).polymorphic_on is\
- BaseType.__table__.c.type
+ assert class_mapper(Specific).polymorphic_on \
+ is BaseType.__table__.c.type
eq_(class_mapper(Specific).polymorphic_identity, 'specific')
def test_non_propagating_mixin_used_for_joined(self):
class TableNameMixin:
+
@classproperty
def __tablename__(cls):
- if (decl.has_inherited_table(cls) and
- TableNameMixin not in cls.__bases__):
+ if decl.has_inherited_table(cls) and TableNameMixin \
+ not in cls.__bases__:
return None
return cls.__name__.lower()
class BaseType(Base, TableNameMixin):
+
discriminator = Column('type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
- id = Column(Integer, primary_key=True)
- value = Column(Integer())
+ __mapper_args__ = dict(polymorphic_on=discriminator)
+ id = Column(Integer, primary_key=True)
+ value = Column(Integer())
class Specific(BaseType, TableNameMixin):
+
__mapper_args__ = dict(polymorphic_identity='specific')
- id = Column(Integer, ForeignKey('basetype.id'), primary_key=True)
-
- eq_(BaseType.__table__.name,'basetype')
- eq_(BaseType.__table__.c.keys(),['type', 'id', 'value'])
- eq_(Specific.__table__.name,'specific')
- eq_(Specific.__table__.c.keys(),['id'])
+ id = Column(Integer, ForeignKey('basetype.id'),
+ primary_key=True)
+
+ eq_(BaseType.__table__.name, 'basetype')
+ eq_(BaseType.__table__.c.keys(), ['type', 'id', 'value'])
+ eq_(Specific.__table__.name, 'specific')
+ eq_(Specific.__table__.c.keys(), ['id'])
def test_single_back_propagate(self):
class ColumnMixin:
- timestamp = Column(Integer)
+
+ timestamp = Column(Integer)
class BaseType(Base):
+
__tablename__ = 'foo'
discriminator = Column('type', String(50))
- __mapper_args__= dict(polymorphic_on=discriminator)
- id = Column(Integer, primary_key=True)
+ __mapper_args__ = dict(polymorphic_on=discriminator)
+ id = Column(Integer, primary_key=True)
+
+ class Specific(BaseType, ColumnMixin):
- class Specific(BaseType,ColumnMixin):
__mapper_args__ = dict(polymorphic_identity='specific')
-
- eq_(BaseType.__table__.c.keys(),['type', 'id', 'timestamp'])
+
+ eq_(BaseType.__table__.c.keys(), ['type', 'id', 'timestamp'])
def test_table_in_model_and_same_column_in_mixin(self):
-
+
class ColumnMixin:
+
data = Column(Integer)
-
+
class Model(Base, ColumnMixin):
- __table__ = Table(
- 'foo',
- Base.metadata,
- Column('data', Integer),
- Column('id', Integer, primary_key=True)
- )
-
+
+ __table__ = Table('foo', Base.metadata, Column('data',
+ Integer), Column('id', Integer,
+ primary_key=True))
+
model_col = Model.__table__.c.data
mixin_col = ColumnMixin.data
assert model_col is not mixin_col
eq_(model_col.name, 'data')
assert model_col.type.__class__ is mixin_col.type.__class__
-
+
def test_table_in_model_and_different_named_column_in_mixin(self):
-
+
class ColumnMixin:
+
tada = Column(Integer)
-
+
def go():
+
class Model(Base, ColumnMixin):
- __table__ = Table(
- 'foo',
- Base.metadata,
- Column('data', Integer),
- Column('id', Integer, primary_key=True)
- )
-
- assert_raises_message(
- sa.exc.ArgumentError,
- "Can't add additional column 'tada' when specifying __table__",
- go)
-
+
+ __table__ = Table('foo', Base.metadata, Column('data',
+ Integer), Column('id', Integer,
+ primary_key=True))
+
+ assert_raises_message(sa.exc.ArgumentError,
+ "Can't add additional column 'tada' when "
+ "specifying __table__", go)
+
def test_table_in_model_overrides_different_typed_column_in_mixin(self):
-
+
class ColumnMixin:
+
data = Column(String)
-
+
class Model(Base, ColumnMixin):
- __table__ = Table(
- 'foo',
- Base.metadata,
- Column('data', Integer),
- Column('id', Integer, primary_key=True)
- )
+
+ __table__ = Table('foo', Base.metadata, Column('data',
+ Integer), Column('id', Integer,
+ primary_key=True))
model_col = Model.__table__.c.data
mixin_col = ColumnMixin.data
assert model_col.type.__class__ is Integer
def test_mixin_column_ordering(self):
-
+
class Foo(object):
+
col1 = Column(Integer)
col3 = Column(Integer)
-
+
class Bar(object):
+
col2 = Column(Integer)
col4 = Column(Integer)
-
+
class Model(Base, Foo, Bar):
+
id = Column(Integer, primary_key=True)
__tablename__ = 'model'
- eq_(Model.__table__.c.keys(),
- ['col1', 'col3', 'col2', 'col4', 'id'])
+ eq_(Model.__table__.c.keys(), ['col1', 'col3', 'col2', 'col4',
+ 'id'])
class DeclarativeMixinPropertyTest(DeclarativeTestBase):
+
def test_column_property(self):
+
class MyMixin(object):
+
@classproperty
def prop_hoho(cls):
return column_property(Column('prop', String(50)))
- class MyModel(Base,MyMixin):
+ class MyModel(Base, MyMixin):
+
__tablename__ = 'test'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
+
+ class MyOtherModel(Base, MyMixin):
- class MyOtherModel(Base,MyMixin):
__tablename__ = 'othertest'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
-
- assert MyModel.__table__.c.prop is not None
- assert MyOtherModel.__table__.c.prop is not None
- assert MyModel.__table__.c.prop is not MyOtherModel.__table__.c.prop
-
- assert MyModel.prop_hoho.property.columns == \
- [MyModel.__table__.c.prop]
- assert MyOtherModel.prop_hoho.property.columns == \
- [MyOtherModel.__table__.c.prop]
- assert MyModel.prop_hoho.property is not \
- MyOtherModel.prop_hoho.property
-
+ test_needs_autoincrement=True)
+
+ assert MyModel.__table__.c.prop is not None
+ assert MyOtherModel.__table__.c.prop is not None
+ assert MyModel.__table__.c.prop \
+ is not MyOtherModel.__table__.c.prop
+ assert MyModel.prop_hoho.property.columns \
+ == [MyModel.__table__.c.prop]
+ assert MyOtherModel.prop_hoho.property.columns \
+ == [MyOtherModel.__table__.c.prop]
+ assert MyModel.prop_hoho.property \
+ is not MyOtherModel.prop_hoho.property
Base.metadata.create_all()
sess = create_session()
m1, m2 = MyModel(prop_hoho='foo'), MyOtherModel(prop_hoho='bar')
sess.add_all([m1, m2])
sess.flush()
- eq_(
- sess.query(MyModel).filter(MyModel.prop_hoho=='foo').one(),
- m1
- )
- eq_(
- sess.query(MyOtherModel).\
- filter(MyOtherModel.prop_hoho=='bar').one(),
- m2
- )
-
+ eq_(sess.query(MyModel).filter(MyModel.prop_hoho == 'foo'
+ ).one(), m1)
+ eq_(sess.query(MyOtherModel).filter(MyOtherModel.prop_hoho
+ == 'bar').one(), m2)
+
def test_doc(self):
"""test documentation transfer.
- the documentation situation with @classproperty is
- problematic. at least see if mapped subclasses
- get the doc.
+ the documentation situation with @classproperty is problematic.
+ at least see if mapped subclasses get the doc.
"""
class MyMixin(object):
+
@classproperty
def type_(cls):
"""this is a document."""
+
return Column(String(50))
@classproperty
def t2(cls):
"""this is another document."""
+
return column_property(Column(String(50)))
-
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
+
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
+
compile_mappers()
- eq_(
- MyModel.type_.__doc__,
- 'this is a document.'
- )
- eq_(
- MyModel.t2.__doc__,
- 'this is another document.'
- )
-
+ eq_(MyModel.type_.__doc__, """this is a document.""")
+ eq_(MyModel.t2.__doc__, """this is another document.""")
+
def test_column_in_mapper_args(self):
+
class MyMixin(object):
+
@classproperty
def type_(cls):
return Column(String(50))
+ __mapper_args__ = {'polymorphic_on': type_}
- __mapper_args__= {'polymorphic_on':type_}
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True)
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True)
-
compile_mappers()
col = MyModel.__mapper__.polymorphic_on
- eq_(col.name,'type_')
+ eq_(col.name, 'type_')
assert col.table is not None
-
+
def test_deferred(self):
+
class MyMixin(object):
+
@classproperty
def data(cls):
return deferred(Column('data', String(50)))
-
- class MyModel(Base,MyMixin):
- __tablename__='test'
- id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
-
+
+ class MyModel(Base, MyMixin):
+
+ __tablename__ = 'test'
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
Base.metadata.create_all()
sess = create_session()
sess.add_all([MyModel(data='d1'), MyModel(data='d2')])
sess.flush()
sess.expunge_all()
-
d1, d2 = sess.query(MyModel).order_by(MyModel.data)
assert 'data' not in d1.__dict__
assert d1.data == 'd1'
assert 'data' in d1.__dict__
-
+
def _test_relationship(self, usestring):
+
class RefTargetMixin(object):
+
@classproperty
def target_id(cls):
return Column('target_id', ForeignKey('target.id'))
-
if usestring:
+
@classproperty
def target(cls):
- return relationship("Target",
- primaryjoin="Target.id==%s.target_id" % cls.__name__
- )
+ return relationship('Target',
+ primaryjoin='Target.id==%s.target_id'
+ % cls.__name__)
else:
+
@classproperty
def target(cls):
- return relationship("Target")
-
+ return relationship('Target')
+
class Foo(Base, RefTargetMixin):
+
__tablename__ = 'foo'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
-
+ test_needs_autoincrement=True)
+
class Bar(Base, RefTargetMixin):
+
__tablename__ = 'bar'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
-
+ test_needs_autoincrement=True)
+
class Target(Base):
+
__tablename__ = 'target'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
-
+ test_needs_autoincrement=True)
+
Base.metadata.create_all()
sess = create_session()
t1, t2 = Target(), Target()
f1, f2, b1 = Foo(target=t1), Foo(target=t2), Bar(target=t1)
sess.add_all([f1, f2, b1])
sess.flush()
-
- eq_(
- sess.query(Foo).filter(Foo.target==t2).one(),
- f2
- )
- eq_(
- sess.query(Bar).filter(Bar.target==t2).first(),
- None
- )
+ eq_(sess.query(Foo).filter(Foo.target == t2).one(), f2)
+ eq_(sess.query(Bar).filter(Bar.target == t2).first(), None)
sess.expire_all()
eq_(f1.target, t1)
-
-
-
+
def test_relationship(self):
self._test_relationship(False)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
