from sqlalchemy.testing import eq_, assert_raises, \
- assert_raises_message, is_
+ assert_raises_message
from sqlalchemy.ext import declarative as decl
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import relationship, create_session, class_mapper, \
joinedload, configure_mappers, backref, clear_mappers, \
- deferred, column_property, composite,\
- Session, properties
-from sqlalchemy.testing import eq_
-from sqlalchemy.util import classproperty, with_metaclass
-from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \
- ConcreteBase, synonym_for
+ column_property, composite, Session, properties
+from sqlalchemy.util import with_metaclass
+from sqlalchemy.ext.declarative import declared_attr, synonym_for
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing.util import gc_collect
Base = None
+User = Address = None
+
+
class DeclarativeTestBase(fixtures.TestBase,
- testing.AssertsExecutionResults,
- testing.AssertsCompiledSQL):
+ testing.AssertsExecutionResults,
+ testing.AssertsCompiledSQL):
__dialect__ = 'default'
+
def setup(self):
global Base
Base = decl.declarative_base(testing.db)
clear_mappers()
Base.metadata.drop_all()
+
class DeclarativeTest(DeclarativeTestBase):
+
def test_basic(self):
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column('name', String(50))
addresses = relationship("Address", backref="user")
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
email = Column(String(50), key='_email')
user_id = Column('user_id', Integer, ForeignKey('users.id'),
key='_user_id')
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column('name', String(50))
addresses = relationship(util.u("Address"), backref="user")
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
email = Column(String(50), key='_email')
user_id = Column('user_id', Integer, ForeignKey('users.id'),
key='_user_id')
__table_args__ = ()
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):
go
)
-
def test_column_repeated_under_prop(self):
def go():
class Foo(Base):
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
+
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
+
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
# metaclass to mock the way zope.interface breaks getattr()
class BrokenMeta(type):
+
def __getattribute__(self, attr):
if attr == 'xyzzy':
raise AttributeError('xyzzy')
else:
- return object.__getattribute__(self,attr)
+ return object.__getattribute__(self, attr)
# even though this class has an xyzzy attribute, getattr(cls,"xyzzy")
# fails
xyzzy = "magic"
# _as_declarative() inspects obj.__class__.__bases__
- class User(BrokenParent,fixtures.ComparableEntity):
+ class User(BrokenParent, fixtures.ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column('name', String(50))
- decl.instrument_declarative(User,{},Base.metadata)
+ decl.instrument_declarative(User, {}, Base.metadata)
def test_reserved_identifiers(self):
def go1():
email = Column('email', String(50))
user_id = Column('user_id', Integer, ForeignKey('users.id'))
user = relationship("User", primaryjoin=user_id == User.id,
- backref="addresses")
+ backref="addresses")
assert mapperlib.Mapper._new_mappers is True
- u = User()
+ u = User() # noqa
assert User.addresses
assert mapperlib.Mapper._new_mappers is False
def test_string_dependency_resolution(self):
- from sqlalchemy.sql import desc
-
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'users'
id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
- addresses = relationship('Address',
- order_by='desc(Address.email)',
- primaryjoin='User.id==Address.user_id',
- foreign_keys='[Address.user_id]',
- backref=backref('user',
- primaryjoin='User.id==Address.user_id',
- foreign_keys='[Address.user_id]'))
+ 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, fixtures.ComparableEntity):
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, fixtures.ComparableEntity):
"ColumnProperty", configure_mappers)
def test_string_dependency_resolution_synonym(self):
- from sqlalchemy.sql import desc
class User(Base, fixtures.ComparableEntity):
id = Column(Integer, primary_key=True)
b_id = Column(ForeignKey('b.id'))
- d = relationship("D",
- secondary="join(B, D, B.d_id == D.id)."
- "join(C, C.d_id == D.id)",
- primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)",
- secondaryjoin="D.id == B.d_id",
- )
+ d = relationship(
+ "D",
+ secondary="join(B, D, B.d_id == D.id)."
+ "join(C, C.d_id == D.id)",
+ primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)",
+ secondaryjoin="D.id == B.d_id",
+ )
class B(Base):
__tablename__ = 'b'
self.assert_compile(
s.query(A).join(A.d),
"SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN "
- "(b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id "
- "JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id "
- "AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id",
+ "(b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id "
+ "JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id "
+ "AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id",
)
def test_string_dependency_resolution_no_table(self):
id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
+
class Address(Base, fixtures.ComparableEntity):
__tablename__ = 'addresses'
test_needs_autoincrement=True)
email = Column(String(50))
user_id = Column(Integer)
- user = relationship("User",
+ user = relationship(
+ "User",
primaryjoin="remote(User.id)==foreign(Address.user_id)"
)
__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, fixtures.ComparableEntity):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
- addresses = relationship('%s.Address' % __name__,
- primaryjoin='%s.User.id==%s.Address.user_id.prop.columns['
- '0]' % (__name__, __name__))
+ addresses = relationship(
+ '%s.Address' % __name__,
+ primaryjoin='%s.User.id==%s.Address.user_id.prop.columns[0]'
+ % (__name__, __name__))
class Address(Base, fixtures.ComparableEntity):
id = Column(Integer, primary_key=True)
name = Column(String(50))
addresses = relationship('Address',
- primaryjoin='User.id==Address.user_id',
- backref='user')
+ primaryjoin='User.id==Address.user_id',
+ backref='user')
class Address(Base, fixtures.ComparableEntity):
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')))
+
configure_mappers()
assert class_mapper(User).get_property('props').secondary \
is user_to_prop
class User(Base):
__tablename__ = 'users'
- __table_args__ = {'schema':'fooschema'}
+ __table_args__ = {'schema': 'fooschema'}
id = Column(Integer, primary_key=True)
name = Column(String(50))
- props = relationship('Prop', secondary='fooschema.user_to_prop',
- primaryjoin='User.id==fooschema.user_to_prop.c.user_id',
- secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id',
- backref='users')
+ props = relationship(
+ 'Prop', secondary='fooschema.user_to_prop',
+ primaryjoin='User.id==fooschema.user_to_prop.c.user_id',
+ secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id',
+ backref='users')
class Prop(Base):
__tablename__ = 'props'
- __table_args__ = {'schema':'fooschema'}
+ __table_args__ = {'schema': 'fooschema'}
id = Column(Integer, primary_key=True)
name = Column(String(50))
- user_to_prop = Table('user_to_prop', Base.metadata,
- Column('user_id', Integer, ForeignKey('fooschema.users.id')),
- Column('prop_id',Integer, ForeignKey('fooschema.props.id')),
- schema='fooschema')
+ user_to_prop = Table(
+ 'user_to_prop', Base.metadata,
+ Column('user_id', Integer, ForeignKey('fooschema.users.id')),
+ Column('prop_id', Integer, ForeignKey('fooschema.props.id')),
+ schema='fooschema')
configure_mappers()
assert class_mapper(User).get_property('props').secondary \
__tablename__ = 'parent'
id = Column(Integer, primary_key=True)
name = Column(String)
- children = relationship("Child",
- primaryjoin="Parent.name==remote(foreign(func.lower(Child.name_upper)))"
- )
+ children = relationship(
+ "Child",
+ primaryjoin="Parent.name=="
+ "remote(foreign(func.lower(Child.name_upper)))"
+ )
class Child(Base):
__tablename__ = 'child'
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)
+ 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
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):
# the exception is preserved. Remains the
# same through repeated calls.
for i in range(3):
- assert_raises_message(sa.exc.InvalidRequestError,
- "^One or more mappers failed to initialize - "
- "can't proceed with initialization of other "
- "mappers. Original exception was: When initializing.*",
- configure_mappers)
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ "^One or more mappers failed to initialize - "
+ "can't proceed with initialization of other "
+ "mappers. Original exception was: When initializing.*",
+ configure_mappers)
def test_custom_base(self):
class MyBase(object):
+
def foobar(self):
return "foobar"
Base = decl.declarative_base(cls=MyBase)
Base.metadata.create_all()
configure_mappers()
assert class_mapper(Detail).get_property('master'
- ).strategy.use_get
+ ).strategy.use_get
m1 = Master()
d1 = Detail(master=m1)
sess = create_session()
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')])
+ 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'))
class ASub(A):
brap = A.data
assert ASub.brap.property is A.data.property
- assert isinstance(ASub.brap.original_property, properties.SynonymProperty)
+ assert isinstance(
+ ASub.brap.original_property, properties.SynonymProperty)
def test_alt_name_attr_subclass_relationship_inline(self):
# [ticket:2900]
id = Column('id', Integer, primary_key=True)
configure_mappers()
+
class ASub(A):
brap = A.b
assert ASub.brap.property is A.b.property
- assert isinstance(ASub.brap.original_property, properties.SynonymProperty)
+ assert isinstance(
+ ASub.brap.original_property, properties.SynonymProperty)
ASub(brap=B())
def test_alt_name_attr_subclass_column_attrset(self):
b_id = Column(Integer, ForeignKey('b.id'))
b = relationship("B", backref="as_")
A.brap = A.b
+
class B(Base):
__tablename__ = 'b'
id = Column('id', Integer, primary_key=True)
assert isinstance(A.brap.original_property, properties.SynonymProperty)
A(brap=B())
-
def test_eager_order_by(self):
class Address(Base, fixtures.ComparableEntity):
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='two'),
- Address(email='one')])
+ 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')])])
+ Address(email='two')])])
def test_order_by_multi(self):
test_needs_autoincrement=True)
name = Column('name', String(50))
addresses = relationship('Address',
- order_by=(Address.email, Address.id))
+ order_by=(Address.email, Address.id))
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='two'),
- Address(email='one')])
+ 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
+ u.addresses
def test_as_declarative(self):
decl.instrument_declarative(Address, reg, Base.metadata)
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ Address(email='two')])
sess = create_session()
sess.add(u1)
sess.flush()
sess.expunge_all()
- eq_(sess.query(User).all(), [User(name='u1',
- addresses=[Address(email='one'), Address(email='two')])])
+ eq_(sess.query(User).all(), [
+ User(
+ name='u1',
+ addresses=[Address(email='one'), Address(email='two')])])
def test_custom_mapper_attribute(self):
__tablename__ = 'foo'
__table_args__ = ForeignKeyConstraint(['id'], ['foo.id'
- ])
+ ])
id = Column('id', Integer, primary_key=True)
assert_raises_message(sa.exc.ArgumentError,
'__table_args__ value must be a tuple, ', err)
User.address_count = \
sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
- where(Address.user_id
- == User.id).as_scalar())
+ where(Address.user_id
+ == User.id).as_scalar())
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ Address(email='two')])
sess = create_session()
sess.add(u1)
sess.flush()
sess.expunge_all()
- eq_(sess.query(User).all(), [User(name='u1', address_count=2,
- addresses=[Address(email='one'), Address(email='two')])])
+ eq_(sess.query(User).all(), [
+ User(name='u1', address_count=2,
+ addresses=[Address(email='one'), Address(email='two')])])
def test_useless_declared_attr(self):
class Address(Base, fixtures.ComparableEntity):
def address_count(cls):
# this doesn't really gain us anything. but if
# one is used, lets have it function as expected...
- return sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
- where(Address.user_id == cls.id))
+ return sa.orm.column_property(
+ sa.select([sa.func.count(Address.id)]).
+ where(Address.user_id == cls.id))
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ Address(email='two')])
sess = create_session()
sess.add(u1)
sess.flush()
sess.expunge_all()
- eq_(sess.query(User).all(), [User(name='u1', address_count=2,
- addresses=[Address(email='one'), Address(email='two')])])
+ eq_(sess.query(User).all(), [
+ User(name='u1', address_count=2,
+ addresses=[Address(email='one'), Address(email='two')])])
def test_declared_on_base_class(self):
class MyBase(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
+
@declared_attr
def somecol(cls):
return Column(Integer)
adr_count = \
sa.orm.column_property(
sa.select([sa.func.count(Address.id)],
- Address.user_id == id).as_scalar())
+ Address.user_id == id).as_scalar())
addresses = relationship(Address)
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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):
eq_(set(User.__table__.c.keys()), set(['id', 'name']))
eq_(set(Address.__table__.c.keys()), set(['id', 'email',
- 'user_id']))
+ 'user_id']))
def test_deferred(self):
def test_composite_inline(self):
class AddressComposite(fixtures.ComparableEntity):
+
def __init__(self, street, state):
self.street = street
self.state = state
+
def __composite_values__(self):
return [self.street, self.state]
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'user'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
address = composite(AddressComposite,
- Column('street', String(50)),
- Column('state', String(2)),
- )
+ Column('street', String(50)),
+ Column('state', String(2)),
+ )
Base.metadata.create_all()
sess = Session()
sess.add(User(
- address=AddressComposite('123 anywhere street',
- 'MD')
- ))
+ address=AddressComposite('123 anywhere street',
+ 'MD')
+ ))
sess.commit()
eq_(
sess.query(User).all(),
[User(address=AddressComposite('123 anywhere street',
- 'MD'))]
+ 'MD'))]
)
def test_composite_separate(self):
class AddressComposite(fixtures.ComparableEntity):
+
def __init__(self, street, state):
self.street = street
self.state = state
+
def __composite_values__(self):
return [self.street, self.state]
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'user'
id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
street = Column(String(50))
state = Column(String(2))
address = composite(AddressComposite,
- street, state)
+ street, state)
Base.metadata.create_all()
sess = Session()
sess.add(User(
- address=AddressComposite('123 anywhere street',
- 'MD')
- ))
+ address=AddressComposite('123 anywhere street',
+ 'MD')
+ ))
sess.commit()
eq_(
sess.query(User).all(),
[User(address=AddressComposite('123 anywhere street',
- 'MD'))]
+ 'MD'))]
)
def test_mapping_to_join(self):
users = Table('users', Base.metadata,
- Column('id', Integer, primary_key=True)
- )
+ Column('id', Integer, primary_key=True)
+ )
addresses = Table('addresses', Base.metadata,
- Column('id', Integer, primary_key=True),
- Column('user_id', Integer, ForeignKey('users.id'))
- )
+ Column('id', Integer, primary_key=True),
+ Column('user_id', Integer, ForeignKey('users.id'))
+ )
usersaddresses = sa.join(users, addresses, users.c.id
== addresses.c.user_id)
+
class User(Base):
__table__ = usersaddresses
- __table_args__ = {'primary_key':[users.c.id]}
+ __table_args__ = {'primary_key': [users.c.id]}
# need to use column_property for now
user_id = column_property(users.c.id, addresses.c.user_id)
address_id = addresses.c.id
assert User.__mapper__.get_property('user_id').columns[0] \
- is users.c.id
+ is users.c.id
assert User.__mapper__.get_property('user_id').columns[1] \
- is addresses.c.user_id
+ is addresses.c.user_id
def test_synonym_inline(self):
name = sa.orm.synonym('_name',
descriptor=property(_get_name,
- _set_name))
+ _set_name))
Base.metadata.create_all()
sess = create_session()
sess.add(u1)
sess.flush()
eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
- ).one(), u1)
+ ).one(), u1)
def test_synonym_no_descriptor(self):
from sqlalchemy.orm.properties import ColumnProperty
sess.add(u1)
sess.flush()
eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
- ).one(), u1)
+ ).one(), u1)
def test_reentrant_compile_via_foreignkey(self):
)
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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):
test_needs_autoincrement=True)
name = Column('name', String(50))
addresses = relationship('Address', backref='user',
- primaryjoin=id == Address.user_id)
+ 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())
+ where(Address.user_id
+ == User.id).as_scalar())
Base.metadata.create_all()
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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):
def test_with_explicit_autoloaded(self):
meta = MetaData(testing.db)
- t1 = Table('t1', meta, Column('id', String(50),
+ t1 = Table(
+ 't1', meta,
+ Column('id', String(50),
primary_key=True, test_needs_autoincrement=True),
- Column('data', String(50)))
+ Column('data', String(50)))
meta.create_all()
try:
sess.add(m)
sess.flush()
eq_(t1.select().execute().fetchall(), [('someid', 'somedata'
- )])
+ )])
finally:
meta.drop_all()
op,
other,
**kw
- ):
+ ):
return op(self.upperself, other, **kw)
class User(Base, fixtures.ComparableEntity):
eq_(rt, u1)
sess.expunge_all()
rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE'
- )).one()
+ )).one()
eq_(rt, u1)
def test_duplicate_classes_in_base(self):
)
-
def _produce_test(inline, stringbased):
class ExplicitJoinTest(fixtures.MappedTest):
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_id, backref='addresses')
if not inline:
configure_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(list(zip(('id', 'name'), column_values)))
- for column_values in [(7, 'jack'), (8, 'ed'), (9,
- 'fred'), (10, 'chuck')]]
+ params = [
+ dict(list(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(list(zip(('id',
- 'user_id', 'email'), column_values)))
- for column_values in [(1, 7, 'jack@bean.com'), (2,
- 8, 'ed@wood.com'), (3, 8, 'ed@bettyboop.com'), (4,
- 8, 'ed@lala.com'), (5, 9, 'fred@fred.com')]])
+ Address.__table__.insert().execute([
+ dict(list(zip(('id', 'user_id', 'email'), column_values)))
+ for column_values in [
+ (1, 7, 'jack@bean.com'),
+ (2, 8, 'ed@wood.com'),
+ (3, 8, 'ed@bettyboop.com'),
+ (4, 8, 'ed@lala.com'), (5, 9, 'fred@fred.com')]])
def test_aliased_join(self):
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')
+ 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:
testclass = _produce_test(inline, stringbased)
exec('%s = testclass' % testclass.__name__)
del testclass
-
from sqlalchemy.ext.declarative import clsregistry
import weakref
+
class MockClass(object):
+
def __init__(self, base, name):
self._decl_class_registry = base
tokens = name.split(".")
f1 = MockClass(base, "foo.bar.Foo")
clsregistry.add_class("Foo", f1)
reg = base['_sa_module_registry']
- mod_entry = reg['foo']['bar']
+ mod_entry = reg['foo']['bar'] # noqa
resolver = clsregistry._resolver(f1, MockProp())
resolver = resolver("foo")
assert_raises_message(
del f4
gc_collect()
assert 'single' not in reg
-
configure_mappers, clear_mappers, \
polymorphic_union, deferred, Session
from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \
- ConcreteBase, has_inherited_table
+ ConcreteBase, has_inherited_table
from sqlalchemy.testing import fixtures
Base = None
+
class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
+
def setup(self):
global Base
Base = decl.declarative_base(testing.db)
clear_mappers()
Base.metadata.drop_all()
+
class DeclarativeInheritanceTest(DeclarativeTestBase):
def test_we_must_copy_mapper_args(self):
assert class_mapper(Person).version_id_col == 'a'
assert class_mapper(Person).include_properties == set(['id', 'a', 'b'])
-
def test_custom_join_condition(self):
class Foo(Base):
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!')])
+ 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')])
+ 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)
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
# ensure that the Manager mapper was compiled with the Manager id
# column as higher priority. this ensures that "Manager.id"
# table (reversed from 0.6's behavior.)
eq_(
- Manager.id.property.columns,
- [Manager.__table__.c.id, Person.__table__.c.id]
+ Manager.id.property.columns,
+ [Manager.__table__.c.id, Person.__table__.c.id]
)
# assert that the "id" column is available without a second
def go():
assert sess.query(Manager).filter(Manager.name == 'dogbert'
- ).one().id
+ ).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
+ ).one().id
self.assert_sql_count(testing.db, go, 1)
primary_key=True)
Engineer.primary_language = Column('primary_language',
- String(50))
+ String(50))
Base.metadata.create_all()
sess = create_session()
e1 = Engineer(primary_language='java', name='dilbert')
sess.flush()
sess.expunge_all()
eq_(sess.query(Person).first(),
- Engineer(primary_language='java', name='dilbert'))
+ Engineer(primary_language='java', name='dilbert'))
def test_add_parentcol_after_the_fact(self):
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 PlanBooking(Booking):
__tablename__ = 'plan_booking'
id = Column(Integer, ForeignKey(Booking.id),
- primary_key=True)
+ primary_key=True)
# referencing PlanBooking.id gives us the column
# on plan_booking, not booking
class FeatureBooking(Booking):
__tablename__ = 'feature_booking'
id = Column(Integer, ForeignKey(Booking.id),
- primary_key=True)
+ primary_key=True)
plan_booking_id = Column(Integer,
- ForeignKey(PlanBooking.id))
+ ForeignKey(PlanBooking.id))
plan_booking = relationship(PlanBooking,
- backref='feature_bookings')
+ backref='feature_bookings')
assert FeatureBooking.__table__.c.plan_booking_id.\
- references(PlanBooking.__table__.c.id)
+ references(PlanBooking.__table__.c.id)
assert FeatureBooking.__table__.c.id.\
- references(Booking.__table__.c.id)
-
+ references(Booking.__table__.c.id)
def test_single_colsonbase(self):
"""test single inheritance where all the columns are on the base
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!')])
+ 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')])
+ 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'))
+ == 'cobol').first(),
+ Engineer(name='vlad'))
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language
- == 'cobol')).first(), c2)
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
def test_single_colsonsub(self):
"""test single inheritance where the columns are local to their
sess.flush()
sess.expunge_all()
eq_(sess.query(Person).filter(Engineer.primary_language
- == 'cobol').first(), Engineer(name='vlad'))
+ == 'cobol').first(),
+ Engineer(name='vlad'))
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language
- == 'cobol')).first(), c2)
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
eq_(sess.query(Engineer).filter_by(primary_language='cobol'
- ).one(), Engineer(name='vlad', primary_language='cobol'))
+ ).one(),
+ Engineer(name='vlad', primary_language='cobol'))
@testing.skip_if(lambda: testing.against('oracle'),
- "Test has an empty insert in it at the moment")
+ "Test has an empty insert in it at the moment")
def test_columns_single_inheritance_conflict_resolution(self):
"""Test that a declared_attr can return the existing column and it will
be ignored. this allows conditional columns to be added.
id = Column(Integer, primary_key=True)
class Engineer(Person):
+
"""single table inheritance"""
@declared_attr
def target_id(cls):
- return cls.__table__.c.get('target_id',
- Column(Integer, ForeignKey('other.id'))
- )
+ return cls.__table__.c.get(
+ 'target_id',
+ Column(Integer, ForeignKey('other.id')))
+
@declared_attr
def target(cls):
return relationship("Other")
class Manager(Person):
+
"""single table inheritance"""
@declared_attr
def target_id(cls):
- return cls.__table__.c.get('target_id',
- Column(Integer, ForeignKey('other.id'))
- )
+ return cls.__table__.c.get(
+ 'target_id',
+ Column(Integer, ForeignKey('other.id')))
+
@declared_attr
def target(cls):
return relationship("Other")
Engineer(target=o1),
Manager(target=o2),
Manager(target=o1)
- ])
+ ])
session.commit()
eq_(session.query(Engineer).first().target, o1)
-
def test_joined_from_single(self):
class Company(Base, fixtures.ComparableEntity):
sess.expunge_all()
eq_(sess.query(Person).with_polymorphic(Engineer).
filter(Engineer.primary_language
- == 'cobol').first(), Engineer(name='vlad'))
+ == 'cobol').first(), Engineer(name='vlad'))
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
- any(Engineer.primary_language
- == 'cobol')).first(), c2)
+ any(Engineer.primary_language
+ == 'cobol')).first(), c2)
eq_(sess.query(Engineer).filter_by(primary_language='cobol'
- ).one(), Engineer(name='vlad', primary_language='cobol'))
+ ).one(),
+ Engineer(name='vlad', primary_language='cobol'))
def test_single_from_joined_colsonsub(self):
class Person(Base, fixtures.ComparableEntity):
eq_(sess.query(Person).all(), [Person(name='ratbert')])
sess.expunge_all()
person = sess.query(Person).filter(Person.name == 'ratbert'
- ).one()
+ ).one()
assert 'name' not in person.__dict__
def test_single_fksonsub(self):
__mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language_id = Column(Integer,
- ForeignKey('languages.id'))
+ ForeignKey('languages.id'))
primary_language = relationship('Language')
class Language(Base, fixtures.ComparableEntity):
sess.expunge_all()
eq_(sess.query(Person).filter(Engineer.primary_language.has(
Language.name
- == 'cobol')).first(), Engineer(name='vlad',
- primary_language=Language(name='cobol')))
+ == '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')))
+ == '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'))])
+ 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):
__mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column('primary_language',
- String(50))
+ String(50))
foo_bar = Column(Integer, primary_key=True)
assert_raises_message(sa.exc.ArgumentError,
- 'place primary key', go)
+ 'place primary key', go)
def test_single_no_table_args(self):
__mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column('primary_language',
- String(50))
+ String(50))
# this should be on the Person class, as this is single
# table inheritance, which is why we test that this
__tablename__ = "a"
id = Column(Integer, primary_key=True)
a_1 = A
+
class A(a_1):
__tablename__ = 'b'
id = Column(Integer(), ForeignKey(a_1.id), primary_key=True)
class OverlapColPrecedenceTest(DeclarativeTestBase):
+
"""test #1892 cases when declarative does column precedence."""
def _run_test(self, Engineer, e_id, p_id):
class Engineer(Person):
__tablename__ = 'engineer'
id = Column("eid", Integer, ForeignKey('person.id'),
- primary_key=True)
+ primary_key=True)
self._run_test(Engineer, "eid", "id")
class Engineer(Person):
__tablename__ = 'engineer'
id = Column("eid", Integer, ForeignKey('person.pid'),
- primary_key=True)
+ primary_key=True)
self._run_test(Engineer, "eid", "pid")
from test.orm.test_events import _RemoveListeners
+
+
class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase):
+
def _roundtrip(self, Employee, Manager, Engineer, Boss,
- polymorphic=True, explicit_type=False):
+ polymorphic=True, explicit_type=False):
Base.metadata.create_all()
sess = create_session()
e1 = Engineer(name='dilbert', primary_language='java')
assert_raises_message(
AttributeError,
"does not implement attribute .?'type' "
- "at the instance level.",
+ "at the instance level.",
getattr, obj, "type"
)
else:
if polymorphic:
eq_(sess.query(Employee).order_by(Employee.name).all(),
[Engineer(name='dilbert'), Manager(name='dogbert'),
- Boss(name='pointy haired'), Engineer(name='vlad'), Engineer(name='wally')])
+ Boss(name='pointy haired'),
+ Engineer(name='vlad'), Engineer(name='wally')])
else:
eq_(sess.query(Engineer).order_by(Engineer.name).all(),
[Engineer(name='dilbert'), Engineer(name='vlad'),
- Engineer(name='wally')])
+ Engineer(name='wally')])
eq_(sess.query(Manager).all(), [Manager(name='dogbert')])
eq_(sess.query(Boss).all(), [Boss(name='pointy haired')])
-
def test_explicit(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)))
+ 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))
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('name', String(50)),
+ Column('golf_swing', String(50))
+ )
boss = Table('boss', Base.metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('name', String(50)),
- Column('golf_swing', String(50))
- )
+ 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,
- 'boss': boss}, 'type', 'punion')
+ 'engineer': engineers,
+ 'manager': managers,
+ 'boss': boss}, 'type', 'punion')
class Employee(Base, fixtures.ComparableEntity):
class Manager(Employee):
__tablename__ = 'manager'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity': 'manager',
- 'concrete': True}
+ 'polymorphic_identity': 'manager',
+ 'concrete': True}
class Boss(Manager):
__tablename__ = 'boss'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity': 'boss',
- 'concrete': True}
+ 'polymorphic_identity': 'boss',
+ 'concrete': True}
class Engineer(Employee):
__tablename__ = 'engineer'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
primary_language = Column(String(40))
__mapper_args__ = {'polymorphic_identity': 'engineer',
- 'concrete': True}
+ 'concrete': True}
self._roundtrip(Employee, Manager, Engineer, Boss)
class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
__tablename__ = 'employee'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
__mapper_args__ = {
- 'polymorphic_identity': 'employee',
- 'concrete': True}
+ 'polymorphic_identity': 'employee',
+ 'concrete': True}
+
class Manager(Employee):
__tablename__ = 'manager'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity': 'manager',
- 'concrete': True}
+ 'polymorphic_identity': 'manager',
+ 'concrete': True}
class Boss(Manager):
__tablename__ = 'boss'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity': 'boss',
- 'concrete': True}
+ 'polymorphic_identity': 'boss',
+ 'concrete': True}
class Engineer(Employee):
__tablename__ = 'engineer'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
primary_language = Column(String(40))
__mapper_args__ = {'polymorphic_identity': 'engineer',
- 'concrete': True}
+ 'concrete': True}
self._roundtrip(Employee, Manager, Engineer, Boss)
-
def test_has_inherited_table_doesnt_consider_base(self):
class A(Base):
__tablename__ = 'a'
ret = {
'polymorphic_identity': 'default',
'polymorphic_on': cls.type,
- }
+ }
else:
ret = {'polymorphic_identity': cls.__name__}
return ret
class Manager(Employee):
__tablename__ = 'manager'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
return "manager"
__mapper_args__ = {
- 'polymorphic_identity': "manager",
- 'concrete': True}
+ 'polymorphic_identity': "manager",
+ 'concrete': True}
class Boss(Manager):
__tablename__ = 'boss'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
return "boss"
__mapper_args__ = {
- 'polymorphic_identity': "boss",
- 'concrete': True}
+ 'polymorphic_identity': "boss",
+ 'concrete': True}
class Engineer(Employee):
__tablename__ = 'engineer'
employee_id = Column(Integer, primary_key=True,
- test_needs_autoincrement=True)
+ test_needs_autoincrement=True)
name = Column(String(50))
primary_language = Column(String(40))
def type(self):
return "engineer"
__mapper_args__ = {'polymorphic_identity': "engineer",
- 'concrete': True}
+ 'concrete': True}
self._roundtrip(Employee, Manager, Engineer, Boss, explicit_type=True)
-class ConcreteExtensionConfigTest(_RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase):
+
+class ConcreteExtensionConfigTest(
+ _RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase):
__dialect__ = 'default'
def test_classreg_setup(self):
class A(Base, fixtures.ComparableEntity):
__tablename__ = 'a'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer,
+ primary_key=True, test_needs_autoincrement=True)
data = Column(String(50))
collection = relationship("BC", primaryjoin="BC.a_id == A.id",
- collection_class=set)
+ collection_class=set)
class BC(AbstractConcreteBase, Base, fixtures.ComparableEntity):
pass
class B(BC):
__tablename__ = 'b'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer,
+ primary_key=True, test_needs_autoincrement=True)
a_id = Column(Integer, ForeignKey('a.id'))
data = Column(String(50))
class C(BC):
__tablename__ = 'c'
- id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer,
+ primary_key=True, test_needs_autoincrement=True)
a_id = Column(Integer, ForeignKey('a.id'))
data = Column(String(50))
c_data = Column(String(50))
sess.query(A).join(A.collection),
"SELECT a.id AS a_id, a.data AS a_data FROM a JOIN "
"(SELECT c.id AS id, c.a_id AS a_id, c.data AS data, "
- "c.c_data AS c_data, CAST(NULL AS VARCHAR(50)) AS b_data, "
- "'c' AS type FROM c UNION ALL SELECT b.id AS id, b.a_id AS a_id, "
- "b.data AS data, CAST(NULL AS VARCHAR(50)) AS c_data, "
- "b.b_data AS b_data, 'b' AS type FROM b) AS pjoin ON pjoin.a_id = a.id"
+ "c.c_data AS c_data, CAST(NULL AS VARCHAR(50)) AS b_data, "
+ "'c' AS type FROM c UNION ALL SELECT b.id AS id, b.a_id AS a_id, "
+ "b.data AS data, CAST(NULL AS VARCHAR(50)) AS c_data, "
+ "b.b_data AS b_data, 'b' AS type FROM b) AS pjoin "
+ "ON pjoin.a_id = a.id"
)
Base = None
+
class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
+
def setup(self):
global Base
Base = decl.declarative_base(testing.db)
clear_mappers()
Base.metadata.drop_all()
+
class DeclarativeMixinTest(DeclarativeTestBase):
def test_simple(self):
def test_table_name_inherited(self):
class MyMixin:
+
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
def test_classproperty_still_works(self):
class MyMixin(object):
+
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
def test_table_name_not_inherited(self):
class MyMixin:
+
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
def test_table_name_inheritance_order(self):
class MyMixin1:
+
@declared_attr
def __tablename__(cls):
return cls.__name__.lower() + '1'
class MyMixin2:
+
@declared_attr
def __tablename__(cls):
return cls.__name__.lower() + '2'
def test_table_name_dependent_on_subclass(self):
class MyHistoryMixin:
+
@declared_attr
def __tablename__(cls):
return cls.parent_name + '_changelog'
def test_table_args_inherited_descriptor(self):
class MyMixin:
+
@declared_attr
def __table_args__(cls):
return {'info': cls.__name__}
assert Specific.bar.prop is General.bar.prop
@testing.skip_if(lambda: testing.against('oracle'),
- "Test has an empty insert in it at the moment")
+ "Test has an empty insert in it at the moment")
def test_columns_single_inheritance_conflict_resolution(self):
"""Test that a declared_attr can return the existing column and it will
be ignored. this allows conditional columns to be added.
id = Column(Integer, primary_key=True)
class Mixin(object):
+
@declared_attr
def target_id(cls):
- return cls.__table__.c.get('target_id',
- Column(Integer, ForeignKey('other.id'))
- )
+ return cls.__table__.c.get(
+ 'target_id',
+ Column(Integer, ForeignKey('other.id'))
+ )
@declared_attr
def target(cls):
return relationship("Other")
class Engineer(Mixin, Person):
+
"""single table inheritance"""
class Manager(Mixin, Person):
+
"""single table inheritance"""
class Other(Base):
Engineer(target=o1),
Manager(target=o2),
Manager(target=o1)
- ])
+ ])
session.commit()
eq_(session.query(Engineer).first().target, o1)
-
def test_columns_joined_table_inheritance(self):
"""Test a column on a mixin with an alternate attribute name,
mapped to a superclass and joined-table inheritance subclass.
def test_mapper_args_declared_attr(self):
class ComputedMapperArgs:
+
@declared_attr
def __mapper_args__(cls):
if cls.__name__ == 'Person':
# ComputedMapperArgs on both classes for no apparent reason.
class ComputedMapperArgs:
+
@declared_attr
def __mapper_args__(cls):
if cls.__name__ == 'Person':
@declared_attr
def __table_args__(cls):
- return {'mysql_engine':'InnoDB'}
+ return {'mysql_engine': 'InnoDB'}
@declared_attr
def __mapper_args__(cls):
"""test the @declared_attr approach from a custom base."""
class Base(object):
+
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
@declared_attr
def __table_args__(cls):
- return {'mysql_engine':'InnoDB'}
+ return {'mysql_engine': 'InnoDB'}
@declared_attr
def id(self):
eq_(Generic.__table__.name, 'generic')
eq_(Specific.__table__.name, 'specific')
eq_(list(Generic.__table__.c.keys()), ['timestamp', 'id',
- 'python_type'])
+ 'python_type'])
eq_(list(Specific.__table__.c.keys()), ['id'])
eq_(Generic.__table__.kwargs, {'mysql_engine': 'InnoDB'})
eq_(Specific.__table__.kwargs, {'mysql_engine': 'InnoDB'})
eq_(BaseType.__table__.name, 'basetype')
eq_(list(BaseType.__table__.c.keys()), ['timestamp', 'type', 'id',
- 'value'])
+ 'value'])
eq_(BaseType.__table__.kwargs, {'mysql_engine': 'InnoDB'})
assert Single.__table__ is BaseType.__table__
eq_(Joined.__table__.name, 'joined')
@declared_attr
def __tablename__(cls):
if decl.has_inherited_table(cls) and TableNameMixin \
- not in cls.__bases__:
+ not in cls.__bases__:
return None
return cls.__name__.lower()
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
class Model(Base, ColumnMixin):
__table__ = Table('foo', Base.metadata,
- Column('data',Integer),
- Column('id', Integer,primary_key=True))
+ Column('data', Integer),
+ Column('id', Integer, primary_key=True))
foo = relationship("Dest")
assert_raises_message(sa.exc.ArgumentError,
class Model(Base, ColumnMixin):
__table__ = Table('foo', Base.metadata,
- Column('data',Integer),
- Column('tada', Integer),
- Column('id', Integer,primary_key=True))
+ Column('data', Integer),
+ Column('tada', Integer),
+ Column('id', Integer, primary_key=True))
foo = relationship("Dest")
assert_raises_message(sa.exc.ArgumentError,
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
__tablename__ = 'model'
eq_(list(Model.__table__.c.keys()), ['col1', 'col3', 'col2', 'col4',
- 'id'])
+ 'id'])
def test_honor_class_mro_one(self):
class HasXMixin(object):
+
@declared_attr
def x(self):
return Column(Integer)
def test_honor_class_mro_two(self):
class HasXMixin(object):
+
@declared_attr
def x(self):
return Column(Integer)
class Parent(HasXMixin, Base):
__tablename__ = 'parent'
id = Column(Integer, primary_key=True)
+
def x(self):
return "hi"
def test_arbitrary_attrs_one(self):
class HasMixin(object):
+
@declared_attr
def some_attr(cls):
return cls.__name__ + "SOME ATTR"
__tablename__ = 'filter_a'
id = Column(Integer(), primary_key=True)
parent_id = Column(Integer(),
- ForeignKey('type_a.id'))
+ ForeignKey('type_a.id'))
filter = Column(String())
+
def __init__(self, filter_, **kw):
self.filter = filter_
__tablename__ = 'filter_b'
id = Column(Integer(), primary_key=True)
parent_id = Column(Integer(),
- ForeignKey('type_b.id'))
+ ForeignKey('type_b.id'))
filter = Column(String())
+
def __init__(self, filter_, **kw):
self.filter = filter_
class FilterMixin(object):
+
@declared_attr
def _filters(cls):
return relationship(cls.filter_class,
- cascade='all,delete,delete-orphan')
+ cascade='all,delete,delete-orphan')
@declared_attr
def filters(cls):
TypeA(filters=['foo'])
TypeB(filters=['foo'])
+
class DeclarativeMixinPropertyTest(DeclarativeTestBase):
def test_column_property(self):
sess.add_all([m1, m2])
sess.flush()
eq_(sess.query(MyModel).filter(MyModel.prop_hoho == 'foo'
- ).one(), m1)
+ ).one(), m1)
eq_(sess.query(MyOtherModel).filter(MyOtherModel.prop_hoho
- == 'bar').one(), m2)
+ == 'bar').one(), m2)
def test_doc(self):
"""test documentation transfer.
ModelTwo.__table__.c.version_id
)
-
def test_deferred(self):
class MyMixin(object):
@declared_attr
def target(cls):
return relationship('Target',
- primaryjoin='Target.id==%s.target_id'
- % cls.__name__)
+ primaryjoin='Target.id==%s.target_id'
+ % cls.__name__)
else:
@declared_attr
class AbstractTest(DeclarativeTestBase):
+
def test_abstract_boolean(self):
class A(Base):
from sqlalchemy.testing import eq_, assert_raises
from sqlalchemy.ext import declarative as decl
from sqlalchemy import testing
-from sqlalchemy import MetaData, Integer, String, ForeignKey
+from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import relationship, create_session, \
clear_mappers, \
from sqlalchemy.testing.util import gc_collect
from sqlalchemy.ext.declarative.base import _DeferredMapperConfig
+
class DeclarativeReflectionBase(fixtures.TablesTest):
__requires__ = 'reflectable_autoincrement',
super(DeclarativeReflectionBase, self).teardown()
clear_mappers()
+
class DeclarativeReflectionTest(DeclarativeReflectionBase):
@classmethod
def define_tables(cls, metadata):
Table('users', metadata,
- Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
+ Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)), test_needs_fk=True)
Table(
'addresses',
Column('email', String(50)),
Column('user_id', Integer, ForeignKey('users.id')),
test_needs_fk=True,
- )
+ )
Table(
'imhandles',
metadata,
Column('network', String(50)),
Column('handle', String(50)),
test_needs_fk=True,
- )
-
+ )
def test_basic(self):
class User(Base, fixtures.ComparableEntity):
test_needs_autoincrement=True)
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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'))
test_needs_autoincrement=True)
u1 = User(nom='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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'))
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')])])
+ 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()
+ ).one()
eq_(a1, IMHandle(network='lol', handle='zomg'))
eq_(a1.user, User(name='u1'))
+
class DeferredReflectBase(DeclarativeReflectionBase):
+
def teardown(self):
super(DeferredReflectBase, self).teardown()
_DeferredMapperConfig._configs.clear()
Base = None
+
class DeferredReflectPKFKTest(DeferredReflectBase):
+
@classmethod
def define_tables(cls, metadata):
Table("a", metadata,
- Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
- )
+ Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ )
Table("b", metadata,
- Column('id', Integer,
- ForeignKey('a.id'),
- primary_key=True),
- Column('x', Integer, primary_key=True)
- )
+ Column('id', Integer,
+ ForeignKey('a.id'),
+ primary_key=True),
+ Column('x', Integer, primary_key=True)
+ )
def test_pk_fk(self):
class B(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'b'
a = relationship("A")
class A(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'a'
decl.DeferredReflection.prepare(testing.db)
+
class DeferredReflectionTest(DeferredReflectBase):
@classmethod
def define_tables(cls, metadata):
Table('users', metadata,
- Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
+ Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)), test_needs_fk=True)
Table(
'addresses',
Column('email', String(50)),
Column('user_id', Integer, ForeignKey('users.id')),
test_needs_fk=True,
- )
+ )
def _roundtrip(self):
Address = Base._decl_class_registry['Address']
u1 = User(name='u1', addresses=[Address(email='one'),
- Address(email='two')])
+ 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_basic_deferred(self):
class User(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'users'
addresses = relationship("Address", backref="user")
class Address(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'addresses'
decl.DeferredReflection.prepare(testing.db)
def test_redefine_fk_double(self):
class User(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'users'
addresses = relationship("Address", backref="user")
class Address(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'addresses'
user_id = Column(Integer, ForeignKey('users.id'))
self._roundtrip()
def test_mapper_args_deferred(self):
- """test that __mapper_args__ is not called until *after* table reflection"""
+ """test that __mapper_args__ is not called until *after*
+ table reflection"""
class User(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'users'
@decl.declared_attr
@testing.requires.predictable_gc
def test_cls_not_strong_ref(self):
class User(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'users'
+
class Address(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'addresses'
eq_(len(_DeferredMapperConfig._configs), 2)
del Address
decl.DeferredReflection.prepare(testing.db)
assert not _DeferredMapperConfig._configs
+
class DeferredSecondaryReflectionTest(DeferredReflectBase):
+
@classmethod
def define_tables(cls, metadata):
Table('users', metadata,
- Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
+ Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
Column('name', String(50)), test_needs_fk=True)
Table('user_items', metadata,
- Column('user_id', ForeignKey('users.id'), primary_key=True),
- Column('item_id', ForeignKey('items.id'), primary_key=True),
- test_needs_fk=True
- )
+ Column('user_id', ForeignKey('users.id'), primary_key=True),
+ Column('item_id', ForeignKey('items.id'), primary_key=True),
+ test_needs_fk=True
+ )
Table('items', metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('name', String(50)),
- test_needs_fk=True
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('name', String(50)),
+ test_needs_fk=True
+ )
def _roundtrip(self):
sess.add(u1)
sess.commit()
- eq_(sess.query(User).all(), [User(name='u1',
- items=[Item(name='i1'), Item(name='i2')])])
+ eq_(sess.query(User).all(), [
+ User(name='u1', items=[Item(name='i1'), Item(name='i2')])])
def test_string_resolution(self):
class User(decl.DeferredReflection, fixtures.ComparableEntity, Base):
class User(decl.DeferredReflection, fixtures.ComparableEntity, Base):
__tablename__ = 'users'
- items = relationship("Item", secondary=Table("user_items", Base.metadata))
+ items = relationship("Item",
+ secondary=Table("user_items", Base.metadata))
class Item(decl.DeferredReflection, fixtures.ComparableEntity, Base):
__tablename__ = 'items'
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
+
class DeferredInhReflectBase(DeferredReflectBase):
+
def _roundtrip(self):
Foo = Base._decl_class_registry['Foo']
Bar = Base._decl_class_registry['Bar']
]
)
+
class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
@classmethod
def define_tables(cls, metadata):
Table("foo", metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('type', String(32)),
- Column('data', String(30)),
- Column('bar_data', String(30))
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('type', String(32)),
+ Column('data', String(30)),
+ Column('bar_data', String(30))
+ )
def test_basic(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__mapper_args__ = {"polymorphic_identity": "bar"}
def test_add_subclass_column(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__mapper_args__ = {"polymorphic_identity": "bar"}
def test_add_pk_column(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
id = Column(Integer, primary_key=True)
class Bar(Foo):
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
+
class DeferredJoinedInhReflectionTest(DeferredInhReflectBase):
+
@classmethod
def define_tables(cls, metadata):
Table("foo", metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('type', String(32)),
- Column('data', String(30)),
- test_needs_fk=True,
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('type', String(32)),
+ Column('data', String(30)),
+ test_needs_fk=True,
+ )
Table('bar', metadata,
- Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
- Column('bar_data', String(30)),
- test_needs_fk=True,
- )
+ Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
+ Column('bar_data', String(30)),
+ test_needs_fk=True,
+ )
def test_basic(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
def test_add_subclass_column(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
def test_add_pk_column(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
id = Column(Integer, primary_key=True)
class Bar(Foo):
def test_add_fk_pk_column(self):
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
- Base):
+ Base):
__tablename__ = 'foo'
__mapper_args__ = {"polymorphic_on": "type",
- "polymorphic_identity": "foo"}
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
