« back to all changes in this revision
Viewing changes to test/ext/test_declarative.py
- browse files at revision 34
- compare with another revision
- download diff
- download tarball
- view history from revision 34
- Committer: Package Import Robot
- Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
- Date: 2013-07-06 20:53:52 UTC
- mfrom: (1.4.23) (16.1.17 experimental)
- Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.
[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
Python 3.X
* Use canonical URIs for Vcs-* fields.
[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
Python 3.X
- files added:
- README.dialects.rst
- doc/_sources/changelog
- doc/build/changelog
- doc/changelog
- examples/dogpile_caching
- lib/sqlalchemy/ext/declarative
- lib/sqlalchemy/testing
- lib/sqlalchemy/testing/plugin
- lib/sqlalchemy/testing/suite
- test/dialect/mssql
- test/dialect/mysql
- test/dialect/postgresql
- test/ext/declarative
- files removed:
- CHANGES_PRE_06
- examples/beaker_caching
- examples/declarative_reflection
- lib/sqlalchemy/dialects/access
- lib/sqlalchemy/dialects/maxdb
- test/bootstrap
- test/ex
- test/lib
- files modified:
- CHANGES
added
removed
test/ext/test_declarative.py
1
2
from test.lib.testing import eq_, assert_raises, \
3
assert_raises_message, is_
4
from sqlalchemy.ext import declarative as decl
5
from sqlalchemy import exc
6
import sqlalchemy as sa
7
from test.lib import testing
8
from sqlalchemy import MetaData, Integer, String, ForeignKey, \
9
ForeignKeyConstraint, Index
10
from test.lib.schema import Table, Column
11
from sqlalchemy.orm import relationship, create_session, class_mapper, \
12
joinedload, configure_mappers, backref, clear_mappers, \
13
polymorphic_union, deferred, column_property, composite,\
14
Session
15
from test.lib.testing import eq_
16
from sqlalchemy.util import classproperty
17
from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, ConcreteBase
18
from test.lib import fixtures
19
20
class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
21
def setup(self):
22
global Base
23
Base = decl.declarative_base(testing.db)
24
25
def teardown(self):
26
Session.close_all()
27
clear_mappers()
28
Base.metadata.drop_all()
29
30
class DeclarativeTest(DeclarativeTestBase):
31
def test_basic(self):
32
class User(Base, fixtures.ComparableEntity):
33
__tablename__ = 'users'
34
35
id = Column('id', Integer, primary_key=True,
36
test_needs_autoincrement=True)
37
name = Column('name', String(50))
38
addresses = relationship("Address", backref="user")
39
40
class Address(Base, fixtures.ComparableEntity):
41
__tablename__ = 'addresses'
42
43
id = Column(Integer, primary_key=True,
44
test_needs_autoincrement=True)
45
email = Column(String(50), key='_email')
46
user_id = Column('user_id', Integer, ForeignKey('users.id'),
47
key='_user_id')
48
49
Base.metadata.create_all()
50
51
eq_(Address.__table__.c['id'].name, 'id')
52
eq_(Address.__table__.c['_email'].name, 'email')
53
eq_(Address.__table__.c['_user_id'].name, 'user_id')
54
55
u1 = User(name='u1', addresses=[
56
Address(email='one'),
57
Address(email='two'),
58
])
59
sess = create_session()
60
sess.add(u1)
61
sess.flush()
62
sess.expunge_all()
63
64
eq_(sess.query(User).all(), [User(name='u1', addresses=[
65
Address(email='one'),
66
Address(email='two'),
67
])])
68
69
a1 = sess.query(Address).filter(Address.email == 'two').one()
70
eq_(a1, Address(email='two'))
71
eq_(a1.user, User(name='u1'))
72
73
def test_no_table(self):
74
def go():
75
class User(Base):
76
id = Column('id', Integer, primary_key=True)
77
78
assert_raises_message(sa.exc.InvalidRequestError,
79
'does not have a __table__', go)
80
81
def test_table_args_empty_dict(self):
82
83
class MyModel(Base):
84
__tablename__ = 'test'
85
id = Column(Integer, primary_key=True)
86
__table_args__ = {}
87
88
def test_table_args_empty_tuple(self):
89
90
class MyModel(Base):
91
__tablename__ = 'test'
92
id = Column(Integer, primary_key=True)
93
__table_args__ = ()
94
95
def test_cant_add_columns(self):
96
t = Table('t', Base.metadata, Column('id', Integer,
97
primary_key=True), Column('data', String))
98
99
def go():
100
class User(Base):
101
__table__ = t
102
foo = Column(Integer, primary_key=True)
103
104
# can't specify new columns not already in the table
105
106
assert_raises_message(sa.exc.ArgumentError,
107
"Can't add additional column 'foo' when "
108
"specifying __table__", go)
109
110
# regular re-mapping works tho
111
112
class Bar(Base):
113
__table__ = t
114
some_data = t.c.data
115
116
assert class_mapper(Bar).get_property('some_data').columns[0] \
117
is t.c.data
118
119
def test_difficult_class(self):
120
"""test no getattr() errors with a customized class"""
121
122
# metaclass to mock the way zope.interface breaks getattr()
123
class BrokenMeta(type):
124
def __getattribute__(self, attr):
125
if attr == 'xyzzy':
126
raise AttributeError, 'xyzzy'
127
else:
128
return object.__getattribute__(self,attr)
129
130
# even though this class has an xyzzy attribute, getattr(cls,"xyzzy")
131
# fails
132
class BrokenParent(object):
133
__metaclass__ = BrokenMeta
134
xyzzy = "magic"
135
136
# _as_declarative() inspects obj.__class__.__bases__
137
class User(BrokenParent,fixtures.ComparableEntity):
138
__tablename__ = 'users'
139
id = Column('id', Integer, primary_key=True,
140
test_needs_autoincrement=True)
141
name = Column('name', String(50))
142
143
decl.instrument_declarative(User,{},Base.metadata)
144
145
def test_reserved_identifiers(self):
146
def go1():
147
class User1(Base):
148
__tablename__ = 'user1'
149
id = Column(Integer, primary_key=True)
150
metadata = Column(Integer)
151
152
def go2():
153
class User2(Base):
154
__tablename__ = 'user2'
155
id = Column(Integer, primary_key=True)
156
metadata = relationship("Address")
157
158
for go in (go1, go2):
159
assert_raises_message(
160
exc.InvalidRequestError,
161
"Attribute name 'metadata' is reserved "
162
"for the MetaData instance when using a "
163
"declarative base class.",
164
go
165
)
166
167
def test_undefer_column_name(self):
168
# TODO: not sure if there was an explicit
169
# test for this elsewhere
170
foo = Column(Integer)
171
eq_(str(foo), '(no name)')
172
eq_(foo.key, None)
173
eq_(foo.name, None)
174
decl._undefer_column_name('foo', foo)
175
eq_(str(foo), 'foo')
176
eq_(foo.key, 'foo')
177
eq_(foo.name, 'foo')
178
179
def test_recompile_on_othermapper(self):
180
"""declarative version of the same test in mappers.py"""
181
182
from sqlalchemy.orm import mapperlib
183
184
class User(Base):
185
__tablename__ = 'users'
186
187
id = Column('id', Integer, primary_key=True)
188
name = Column('name', String(50))
189
190
class Address(Base):
191
__tablename__ = 'addresses'
192
193
id = Column('id', Integer, primary_key=True)
194
email = Column('email', String(50))
195
user_id = Column('user_id', Integer, ForeignKey('users.id'))
196
user = relationship("User", primaryjoin=user_id == User.id,
197
backref="addresses")
198
199
assert mapperlib._new_mappers is True
200
u = User()
201
assert User.addresses
202
assert mapperlib._new_mappers is False
203
204
def test_string_dependency_resolution(self):
205
from sqlalchemy.sql import desc
206
207
class User(Base, fixtures.ComparableEntity):
208
209
__tablename__ = 'users'
210
id = Column(Integer, primary_key=True,
211
test_needs_autoincrement=True)
212
name = Column(String(50))
213
addresses = relationship('Address',
214
order_by='desc(Address.email)',
215
primaryjoin='User.id==Address.user_id',
216
foreign_keys='[Address.user_id]',
217
backref=backref('user',
218
primaryjoin='User.id==Address.user_id',
219
foreign_keys='[Address.user_id]'))
220
221
class Address(Base, fixtures.ComparableEntity):
222
223
__tablename__ = 'addresses'
224
id = Column(Integer, primary_key=True,
225
test_needs_autoincrement=True)
226
email = Column(String(50))
227
user_id = Column(Integer) # note no foreign key
228
229
Base.metadata.create_all()
230
sess = create_session()
231
u1 = User(name='ed', addresses=[Address(email='abc'),
232
Address(email='def'), Address(email='xyz')])
233
sess.add(u1)
234
sess.flush()
235
sess.expunge_all()
236
eq_(sess.query(User).filter(User.name == 'ed').one(),
237
User(name='ed', addresses=[Address(email='xyz'),
238
Address(email='def'), Address(email='abc')]))
239
240
class Foo(Base, fixtures.ComparableEntity):
241
242
__tablename__ = 'foo'
243
id = Column(Integer, primary_key=True)
244
rel = relationship('User',
245
primaryjoin='User.addresses==Foo.id')
246
247
assert_raises_message(exc.InvalidRequestError,
248
"'addresses' is not an instance of "
249
"ColumnProperty", configure_mappers)
250
251
def test_string_dependency_resolution_two(self):
252
253
class User(Base, fixtures.ComparableEntity):
254
__tablename__ = 'users'
255
id = Column(Integer, primary_key=True,
256
test_needs_autoincrement=True)
257
name = Column(String(50))
258
259
class Bar(Base, fixtures.ComparableEntity):
260
__tablename__ = 'bar'
261
id = Column(Integer, primary_key=True)
262
rel = relationship('User',
263
primaryjoin='User.id==Bar.__table__.id')
264
265
assert_raises_message(exc.InvalidRequestError,
266
"does not have a mapped column named "
267
"'__table__'", configure_mappers)
268
269
def test_string_dependency_resolution_no_magic(self):
270
"""test that full tinkery expressions work as written"""
271
272
class User(Base, fixtures.ComparableEntity):
273
274
__tablename__ = 'users'
275
id = Column(Integer, primary_key=True)
276
addresses = relationship('Address',
277
primaryjoin='User.id==Address.user_id.prop.columns['
278
'0]')
279
280
class Address(Base, fixtures.ComparableEntity):
281
282
__tablename__ = 'addresses'
283
id = Column(Integer, primary_key=True)
284
user_id = Column(Integer, ForeignKey('users.id'))
285
286
configure_mappers()
287
eq_(str(User.addresses.prop.primaryjoin),
288
'users.id = addresses.user_id')
289
290
def test_string_dependency_resolution_in_backref(self):
291
292
class User(Base, fixtures.ComparableEntity):
293
294
__tablename__ = 'users'
295
id = Column(Integer, primary_key=True)
296
name = Column(String(50))
297
addresses = relationship('Address',
298
primaryjoin='User.id==Address.user_id',
299
backref='user')
300
301
class Address(Base, fixtures.ComparableEntity):
302
303
__tablename__ = 'addresses'
304
id = Column(Integer, primary_key=True)
305
email = Column(String(50))
306
user_id = Column(Integer, ForeignKey('users.id'))
307
308
configure_mappers()
309
eq_(str(User.addresses.property.primaryjoin),
310
str(Address.user.property.primaryjoin))
311
312
def test_string_dependency_resolution_tables(self):
313
314
class User(Base, fixtures.ComparableEntity):
315
316
__tablename__ = 'users'
317
id = Column(Integer, primary_key=True)
318
name = Column(String(50))
319
props = relationship('Prop', secondary='user_to_prop',
320
primaryjoin='User.id==user_to_prop.c.u'
321
'ser_id',
322
secondaryjoin='user_to_prop.c.prop_id='
323
'=Prop.id', backref='users')
324
325
class Prop(Base, fixtures.ComparableEntity):
326
327
__tablename__ = 'props'
328
id = Column(Integer, primary_key=True)
329
name = Column(String(50))
330
331
user_to_prop = Table('user_to_prop', Base.metadata,
332
Column('user_id', Integer,
333
ForeignKey('users.id')), Column('prop_id',
334
Integer, ForeignKey('props.id')))
335
configure_mappers()
336
assert class_mapper(User).get_property('props').secondary \
337
is user_to_prop
338
339
def test_string_dependency_resolution_schemas(self):
340
Base = decl.declarative_base()
341
342
class User(Base):
343
344
__tablename__ = 'users'
345
__table_args__ = {'schema':'fooschema'}
346
347
id = Column(Integer, primary_key=True)
348
name = Column(String(50))
349
props = relationship('Prop', secondary='fooschema.user_to_prop',
350
primaryjoin='User.id==fooschema.user_to_prop.c.user_id',
351
secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id',
352
backref='users')
353
354
class Prop(Base):
355
356
__tablename__ = 'props'
357
__table_args__ = {'schema':'fooschema'}
358
359
id = Column(Integer, primary_key=True)
360
name = Column(String(50))
361
362
user_to_prop = Table('user_to_prop', Base.metadata,
363
Column('user_id', Integer, ForeignKey('fooschema.users.id')),
364
Column('prop_id',Integer, ForeignKey('fooschema.props.id')),
365
schema='fooschema')
366
configure_mappers()
367
368
assert class_mapper(User).get_property('props').secondary \
369
is user_to_prop
370
371
def test_shared_class_registry(self):
372
reg = {}
373
Base1 = decl.declarative_base(testing.db, class_registry=reg)
374
Base2 = decl.declarative_base(testing.db, class_registry=reg)
375
376
class A(Base1):
377
__tablename__ = 'a'
378
id = Column(Integer, primary_key=True)
379
380
class B(Base2):
381
__tablename__ = 'b'
382
id = Column(Integer, primary_key=True)
383
aid = Column(Integer, ForeignKey(A.id))
384
as_ = relationship("A")
385
386
assert B.as_.property.mapper.class_ is A
387
388
def test_uncompiled_attributes_in_relationship(self):
389
390
class Address(Base, fixtures.ComparableEntity):
391
392
__tablename__ = 'addresses'
393
id = Column(Integer, primary_key=True,
394
test_needs_autoincrement=True)
395
email = Column(String(50))
396
user_id = Column(Integer, ForeignKey('users.id'))
397
398
class User(Base, fixtures.ComparableEntity):
399
400
__tablename__ = 'users'
401
id = Column(Integer, primary_key=True,
402
test_needs_autoincrement=True)
403
name = Column(String(50))
404
addresses = relationship('Address', order_by=Address.email,
405
foreign_keys=Address.user_id,
406
remote_side=Address.user_id)
407
408
# get the mapper for User. User mapper will compile,
409
# "addresses" relationship will call upon Address.user_id for
410
# its clause element. Address.user_id is a _CompileOnAttr,
411
# which then calls class_mapper(Address). But ! We're already
412
# "in compilation", but class_mapper(Address) needs to
413
# initialize regardless, or COA's assertion fails and things
414
# generally go downhill from there.
415
416
class_mapper(User)
417
Base.metadata.create_all()
418
sess = create_session()
419
u1 = User(name='ed', addresses=[Address(email='abc'),
420
Address(email='xyz'), Address(email='def')])
421
sess.add(u1)
422
sess.flush()
423
sess.expunge_all()
424
eq_(sess.query(User).filter(User.name == 'ed').one(),
425
User(name='ed', addresses=[Address(email='abc'),
426
Address(email='def'), Address(email='xyz')]))
427
428
def test_nice_dependency_error(self):
429
430
class User(Base):
431
432
__tablename__ = 'users'
433
id = Column('id', Integer, primary_key=True)
434
addresses = relationship('Address')
435
436
class Address(Base):
437
438
__tablename__ = 'addresses'
439
id = Column(Integer, primary_key=True)
440
foo = sa.orm.column_property(User.id == 5)
441
442
# this used to raise an error when accessing User.id but that's
443
# no longer the case since we got rid of _CompileOnAttr.
444
445
assert_raises(sa.exc.ArgumentError, configure_mappers)
446
447
def test_nice_dependency_error_works_with_hasattr(self):
448
449
class User(Base):
450
451
__tablename__ = 'users'
452
id = Column('id', Integer, primary_key=True)
453
addresses = relationship('Address')
454
455
# hasattr() on a compile-loaded attribute
456
try:
457
hasattr(User.addresses, 'property')
458
except exc.InvalidRequestError:
459
assert sa.util.compat.py32
460
461
# the exception is preserved. Remains the
462
# same through repeated calls.
463
for i in range(3):
464
assert_raises_message(sa.exc.InvalidRequestError,
465
"^One or more mappers failed to initialize - "
466
"can't proceed with initialization of other "
467
"mappers. Original exception was: When initializing.*",
468
configure_mappers)
469
470
def test_custom_base(self):
471
class MyBase(object):
472
def foobar(self):
473
return "foobar"
474
Base = decl.declarative_base(cls=MyBase)
475
assert hasattr(Base, 'metadata')
476
assert Base().foobar() == "foobar"
477
478
def test_uses_get_on_class_col_fk(self):
479
480
# test [ticket:1492]
481
482
class Master(Base):
483
484
__tablename__ = 'master'
485
id = Column(Integer, primary_key=True,
486
test_needs_autoincrement=True)
487
488
class Detail(Base):
489
490
__tablename__ = 'detail'
491
id = Column(Integer, primary_key=True,
492
test_needs_autoincrement=True)
493
master_id = Column(None, ForeignKey(Master.id))
494
master = relationship(Master)
495
496
Base.metadata.create_all()
497
configure_mappers()
498
assert class_mapper(Detail).get_property('master'
499
).strategy.use_get
500
m1 = Master()
501
d1 = Detail(master=m1)
502
sess = create_session()
503
sess.add(d1)
504
sess.flush()
505
sess.expunge_all()
506
d1 = sess.query(Detail).first()
507
m1 = sess.query(Master).first()
508
509
def go():
510
assert d1.master
511
512
self.assert_sql_count(testing.db, go, 0)
513
514
def test_index_doesnt_compile(self):
515
class User(Base):
516
__tablename__ = 'users'
517
id = Column('id', Integer, primary_key=True)
518
name = Column('name', String(50))
519
error = relationship("Address")
520
521
i = Index('my_index', User.name)
522
523
# compile fails due to the nonexistent Addresses relationship
524
assert_raises(sa.exc.InvalidRequestError, configure_mappers)
525
526
# index configured
527
assert i in User.__table__.indexes
528
assert User.__table__.c.id not in set(i.columns)
529
assert User.__table__.c.name in set(i.columns)
530
531
# tables create fine
532
Base.metadata.create_all()
533
534
def test_add_prop(self):
535
536
class User(Base, fixtures.ComparableEntity):
537
538
__tablename__ = 'users'
539
id = Column('id', Integer, primary_key=True,
540
test_needs_autoincrement=True)
541
542
User.name = Column('name', String(50))
543
User.addresses = relationship('Address', backref='user')
544
545
class Address(Base, fixtures.ComparableEntity):
546
547
__tablename__ = 'addresses'
548
id = Column(Integer, primary_key=True,
549
test_needs_autoincrement=True)
550
551
Address.email = Column(String(50), key='_email')
552
Address.user_id = Column('user_id', Integer,
553
ForeignKey('users.id'), key='_user_id')
554
Base.metadata.create_all()
555
eq_(Address.__table__.c['id'].name, 'id')
556
eq_(Address.__table__.c['_email'].name, 'email')
557
eq_(Address.__table__.c['_user_id'].name, 'user_id')
558
u1 = User(name='u1', addresses=[Address(email='one'),
559
Address(email='two')])
560
sess = create_session()
561
sess.add(u1)
562
sess.flush()
563
sess.expunge_all()
564
eq_(sess.query(User).all(), [User(name='u1',
565
addresses=[Address(email='one'), Address(email='two')])])
566
a1 = sess.query(Address).filter(Address.email == 'two').one()
567
eq_(a1, Address(email='two'))
568
eq_(a1.user, User(name='u1'))
569
570
def test_eager_order_by(self):
571
572
class Address(Base, fixtures.ComparableEntity):
573
574
__tablename__ = 'addresses'
575
id = Column('id', Integer, primary_key=True,
576
test_needs_autoincrement=True)
577
email = Column('email', String(50))
578
user_id = Column('user_id', Integer, ForeignKey('users.id'))
579
580
class User(Base, fixtures.ComparableEntity):
581
582
__tablename__ = 'users'
583
id = Column('id', Integer, primary_key=True,
584
test_needs_autoincrement=True)
585
name = Column('name', String(50))
586
addresses = relationship('Address', order_by=Address.email)
587
588
Base.metadata.create_all()
589
u1 = User(name='u1', addresses=[Address(email='two'),
590
Address(email='one')])
591
sess = create_session()
592
sess.add(u1)
593
sess.flush()
594
sess.expunge_all()
595
eq_(sess.query(User).options(joinedload(User.addresses)).all(),
596
[User(name='u1', addresses=[Address(email='one'),
597
Address(email='two')])])
598
599
def test_order_by_multi(self):
600
601
class Address(Base, fixtures.ComparableEntity):
602
603
__tablename__ = 'addresses'
604
id = Column('id', Integer, primary_key=True,
605
test_needs_autoincrement=True)
606
email = Column('email', String(50))
607
user_id = Column('user_id', Integer, ForeignKey('users.id'))
608
609
class User(Base, fixtures.ComparableEntity):
610
611
__tablename__ = 'users'
612
id = Column('id', Integer, primary_key=True,
613
test_needs_autoincrement=True)
614
name = Column('name', String(50))
615
addresses = relationship('Address',
616
order_by=(Address.email, Address.id))
617
618
Base.metadata.create_all()
619
u1 = User(name='u1', addresses=[Address(email='two'),
620
Address(email='one')])
621
sess = create_session()
622
sess.add(u1)
623
sess.flush()
624
sess.expunge_all()
625
u = sess.query(User).filter(User.name == 'u1').one()
626
a = u.addresses
627
628
def test_as_declarative(self):
629
630
class User(fixtures.ComparableEntity):
631
632
__tablename__ = 'users'
633
id = Column('id', Integer, primary_key=True,
634
test_needs_autoincrement=True)
635
name = Column('name', String(50))
636
addresses = relationship('Address', backref='user')
637
638
class Address(fixtures.ComparableEntity):
639
640
__tablename__ = 'addresses'
641
id = Column('id', Integer, primary_key=True,
642
test_needs_autoincrement=True)
643
email = Column('email', String(50))
644
user_id = Column('user_id', Integer, ForeignKey('users.id'))
645
646
reg = {}
647
decl.instrument_declarative(User, reg, Base.metadata)
648
decl.instrument_declarative(Address, reg, Base.metadata)
649
Base.metadata.create_all()
650
u1 = User(name='u1', addresses=[Address(email='one'),
651
Address(email='two')])
652
sess = create_session()
653
sess.add(u1)
654
sess.flush()
655
sess.expunge_all()
656
eq_(sess.query(User).all(), [User(name='u1',
657
addresses=[Address(email='one'), Address(email='two')])])
658
659
def test_custom_mapper_attribute(self):
660
661
def mymapper(cls, tbl, **kwargs):
662
m = sa.orm.mapper(cls, tbl, **kwargs)
663
m.CHECK = True
664
return m
665
666
base = decl.declarative_base()
667
668
class Foo(base):
669
__tablename__ = 'foo'
670
__mapper_cls__ = mymapper
671
id = Column(Integer, primary_key=True)
672
673
eq_(Foo.__mapper__.CHECK, True)
674
675
def test_custom_mapper_argument(self):
676
677
def mymapper(cls, tbl, **kwargs):
678
m = sa.orm.mapper(cls, tbl, **kwargs)
679
m.CHECK = True
680
return m
681
682
base = decl.declarative_base(mapper=mymapper)
683
684
class Foo(base):
685
__tablename__ = 'foo'
686
id = Column(Integer, primary_key=True)
687
688
eq_(Foo.__mapper__.CHECK, True)
689
690
@testing.emits_warning('Ignoring declarative-like tuple value of '
691
'attribute id')
692
def test_oops(self):
693
694
def define():
695
696
class User(Base, fixtures.ComparableEntity):
697
698
__tablename__ = 'users'
699
id = Column('id', Integer, primary_key=True),
700
name = Column('name', String(50))
701
702
assert False
703
704
assert_raises_message(sa.exc.ArgumentError,
705
'Mapper Mapper|User|users could not '
706
'assemble any primary key', define)
707
708
def test_table_args_no_dict(self):
709
710
class Foo1(Base):
711
712
__tablename__ = 'foo'
713
__table_args__ = ForeignKeyConstraint(['id'], ['foo.bar']),
714
id = Column('id', Integer, primary_key=True)
715
bar = Column('bar', Integer)
716
717
assert Foo1.__table__.c.id.references(Foo1.__table__.c.bar)
718
719
def test_table_args_type(self):
720
def err():
721
class Foo1(Base):
722
723
__tablename__ = 'foo'
724
__table_args__ = ForeignKeyConstraint(['id'], ['foo.id'
725
])
726
id = Column('id', Integer, primary_key=True)
727
assert_raises_message(sa.exc.ArgumentError,
728
'__table_args__ value must be a tuple, ', err)
729
730
def test_table_args_none(self):
731
732
class Foo2(Base):
733
734
__tablename__ = 'foo'
735
__table_args__ = None
736
id = Column('id', Integer, primary_key=True)
737
738
assert Foo2.__table__.kwargs == {}
739
740
def test_table_args_dict_format(self):
741
742
class Foo2(Base):
743
744
__tablename__ = 'foo'
745
__table_args__ = {'mysql_engine': 'InnoDB'}
746
id = Column('id', Integer, primary_key=True)
747
748
assert Foo2.__table__.kwargs['mysql_engine'] == 'InnoDB'
749
750
def test_table_args_tuple_format(self):
751
class Foo2(Base):
752
753
__tablename__ = 'foo'
754
__table_args__ = {'mysql_engine': 'InnoDB'}
755
id = Column('id', Integer, primary_key=True)
756
757
class Bar(Base):
758
759
__tablename__ = 'bar'
760
__table_args__ = ForeignKeyConstraint(['id'], ['foo.id']), \
761
{'mysql_engine': 'InnoDB'}
762
id = Column('id', Integer, primary_key=True)
763
764
assert Bar.__table__.c.id.references(Foo2.__table__.c.id)
765
assert Bar.__table__.kwargs['mysql_engine'] == 'InnoDB'
766
767
def test_expression(self):
768
769
class User(Base, fixtures.ComparableEntity):
770
771
__tablename__ = 'users'
772
id = Column('id', Integer, primary_key=True,
773
test_needs_autoincrement=True)
774
name = Column('name', String(50))
775
addresses = relationship('Address', backref='user')
776
777
class Address(Base, fixtures.ComparableEntity):
778
779
__tablename__ = 'addresses'
780
id = Column('id', Integer, primary_key=True,
781
test_needs_autoincrement=True)
782
email = Column('email', String(50))
783
user_id = Column('user_id', Integer, ForeignKey('users.id'))
784
785
User.address_count = \
786
sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
787
where(Address.user_id
788
== User.id).as_scalar())
789
Base.metadata.create_all()
790
u1 = User(name='u1', addresses=[Address(email='one'),
791
Address(email='two')])
792
sess = create_session()
793
sess.add(u1)
794
sess.flush()
795
sess.expunge_all()
796
eq_(sess.query(User).all(), [User(name='u1', address_count=2,
797
addresses=[Address(email='one'), Address(email='two')])])
798
799
def test_useless_declared_attr(self):
800
class Address(Base, fixtures.ComparableEntity):
801
802
__tablename__ = 'addresses'
803
id = Column('id', Integer, primary_key=True,
804
test_needs_autoincrement=True)
805
email = Column('email', String(50))
806
user_id = Column('user_id', Integer, ForeignKey('users.id'))
807
808
class User(Base, fixtures.ComparableEntity):
809
810
__tablename__ = 'users'
811
id = Column('id', Integer, primary_key=True,
812
test_needs_autoincrement=True)
813
name = Column('name', String(50))
814
addresses = relationship('Address', backref='user')
815
816
@declared_attr
817
def address_count(cls):
818
# this doesn't really gain us anything. but if
819
# one is used, lets have it function as expected...
820
return sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
821
where(Address.user_id == cls.id))
822
823
Base.metadata.create_all()
824
u1 = User(name='u1', addresses=[Address(email='one'),
825
Address(email='two')])
826
sess = create_session()
827
sess.add(u1)
828
sess.flush()
829
sess.expunge_all()
830
eq_(sess.query(User).all(), [User(name='u1', address_count=2,
831
addresses=[Address(email='one'), Address(email='two')])])
832
833
def test_useless_declared_attr_warns_on_subclass(self):
834
def go():
835
class MyBase(Base):
836
__tablename__ = 'foo'
837
id = Column(Integer, primary_key=True)
838
@declared_attr
839
def somecol(cls):
840
return Column(Integer)
841
842
class MyClass(MyBase):
843
__tablename__ = 'bar'
844
assert_raises_message(
845
sa.exc.SAWarning,
846
r"Regular \(i.e. not __special__\) attribute 'MyBase.somecol' "
847
"uses @declared_attr, but owning class "
848
"<class 'test.ext..*test_declarative..*MyBase'> is "
849
"mapped - not applying to subclass <class "
850
"'test.ext..*test_declarative..*MyClass'>.",
851
go
852
)
853
854
def test_column(self):
855
856
class User(Base, fixtures.ComparableEntity):
857
858
__tablename__ = 'users'
859
id = Column('id', Integer, primary_key=True,
860
test_needs_autoincrement=True)
861
name = Column('name', String(50))
862
863
User.a = Column('a', String(10))
864
User.b = Column(String(10))
865
Base.metadata.create_all()
866
u1 = User(name='u1', a='a', b='b')
867
eq_(u1.a, 'a')
868
eq_(User.a.get_history(u1), (['a'], (), ()))
869
sess = create_session()
870
sess.add(u1)
871
sess.flush()
872
sess.expunge_all()
873
eq_(sess.query(User).all(), [User(name='u1', a='a', b='b')])
874
875
def test_column_properties(self):
876
877
class Address(Base, fixtures.ComparableEntity):
878
879
__tablename__ = 'addresses'
880
id = Column(Integer, primary_key=True,
881
test_needs_autoincrement=True)
882
email = Column(String(50))
883
user_id = Column(Integer, ForeignKey('users.id'))
884
885
class User(Base, fixtures.ComparableEntity):
886
887
__tablename__ = 'users'
888
id = Column('id', Integer, primary_key=True,
889
test_needs_autoincrement=True)
890
name = Column('name', String(50))
891
892
adr_count = \
893
sa.orm.column_property(
894
sa.select([sa.func.count(Address.id)],
895
Address.user_id == id).as_scalar())
896
addresses = relationship(Address)
897
898
Base.metadata.create_all()
899
u1 = User(name='u1', addresses=[Address(email='one'),
900
Address(email='two')])
901
sess = create_session()
902
sess.add(u1)
903
sess.flush()
904
sess.expunge_all()
905
eq_(sess.query(User).all(), [User(name='u1', adr_count=2,
906
addresses=[Address(email='one'), Address(email='two')])])
907
908
def test_column_properties_2(self):
909
910
class Address(Base, fixtures.ComparableEntity):
911
912
__tablename__ = 'addresses'
913
id = Column(Integer, primary_key=True)
914
email = Column(String(50))
915
user_id = Column(Integer, ForeignKey('users.id'))
916
917
class User(Base, fixtures.ComparableEntity):
918
919
__tablename__ = 'users'
920
id = Column('id', Integer, primary_key=True)
921
name = Column('name', String(50))
922
923
# this is not "valid" but we want to test that Address.id
924
# doesnt get stuck into user's table
925
926
adr_count = Address.id
927
928
eq_(set(User.__table__.c.keys()), set(['id', 'name']))
929
eq_(set(Address.__table__.c.keys()), set(['id', 'email',
930
'user_id']))
931
932
def test_deferred(self):
933
934
class User(Base, fixtures.ComparableEntity):
935
936
__tablename__ = 'users'
937
id = Column(Integer, primary_key=True,
938
test_needs_autoincrement=True)
939
name = sa.orm.deferred(Column(String(50)))
940
941
Base.metadata.create_all()
942
sess = create_session()
943
sess.add(User(name='u1'))
944
sess.flush()
945
sess.expunge_all()
946
u1 = sess.query(User).filter(User.name == 'u1').one()
947
assert 'name' not in u1.__dict__
948
949
def go():
950
eq_(u1.name, 'u1')
951
952
self.assert_sql_count(testing.db, go, 1)
953
954
def test_composite_inline(self):
955
class AddressComposite(fixtures.ComparableEntity):
956
def __init__(self, street, state):
957
self.street = street
958
self.state = state
959
def __composite_values__(self):
960
return [self.street, self.state]
961
962
class User(Base, fixtures.ComparableEntity):
963
__tablename__ = 'user'
964
id = Column(Integer, primary_key=True,
965
test_needs_autoincrement=True)
966
address = composite(AddressComposite,
967
Column('street', String(50)),
968
Column('state', String(2)),
969
)
970
971
Base.metadata.create_all()
972
sess = Session()
973
sess.add(User(
974
address=AddressComposite('123 anywhere street',
975
'MD')
976
))
977
sess.commit()
978
eq_(
979
sess.query(User).all(),
980
[User(address=AddressComposite('123 anywhere street',
981
'MD'))]
982
)
983
984
def test_composite_separate(self):
985
class AddressComposite(fixtures.ComparableEntity):
986
def __init__(self, street, state):
987
self.street = street
988
self.state = state
989
def __composite_values__(self):
990
return [self.street, self.state]
991
992
class User(Base, fixtures.ComparableEntity):
993
__tablename__ = 'user'
994
id = Column(Integer, primary_key=True,
995
test_needs_autoincrement=True)
996
street = Column(String(50))
997
state = Column(String(2))
998
address = composite(AddressComposite,
999
street, state)
1000
1001
Base.metadata.create_all()
1002
sess = Session()
1003
sess.add(User(
1004
address=AddressComposite('123 anywhere street',
1005
'MD')
1006
))
1007
sess.commit()
1008
eq_(
1009
sess.query(User).all(),
1010
[User(address=AddressComposite('123 anywhere street',
1011
'MD'))]
1012
)
1013
1014
def test_mapping_to_join(self):
1015
users = Table('users', Base.metadata,
1016
Column('id', Integer, primary_key=True)
1017
)
1018
addresses = Table('addresses', Base.metadata,
1019
Column('id', Integer, primary_key=True),
1020
Column('user_id', Integer, ForeignKey('users.id'))
1021
)
1022
usersaddresses = sa.join(users, addresses, users.c.id
1023
== addresses.c.user_id)
1024
class User(Base):
1025
__table__ = usersaddresses
1026
__table_args__ = {'primary_key':[users.c.id]}
1027
1028
# need to use column_property for now
1029
user_id = column_property(users.c.id, addresses.c.user_id)
1030
address_id = addresses.c.id
1031
1032
assert User.__mapper__.get_property('user_id').columns[0] \
1033
is users.c.id
1034
assert User.__mapper__.get_property('user_id').columns[1] \
1035
is addresses.c.user_id
1036
1037
def test_synonym_inline(self):
1038
1039
class User(Base, fixtures.ComparableEntity):
1040
1041
__tablename__ = 'users'
1042
id = Column('id', Integer, primary_key=True,
1043
test_needs_autoincrement=True)
1044
_name = Column('name', String(50))
1045
1046
def _set_name(self, name):
1047
self._name = 'SOMENAME ' + name
1048
1049
def _get_name(self):
1050
return self._name
1051
1052
name = sa.orm.synonym('_name',
1053
descriptor=property(_get_name,
1054
_set_name))
1055
1056
Base.metadata.create_all()
1057
sess = create_session()
1058
u1 = User(name='someuser')
1059
eq_(u1.name, 'SOMENAME someuser')
1060
sess.add(u1)
1061
sess.flush()
1062
eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
1063
).one(), u1)
1064
1065
def test_synonym_no_descriptor(self):
1066
from sqlalchemy.orm.properties import ColumnProperty
1067
1068
class CustomCompare(ColumnProperty.Comparator):
1069
1070
__hash__ = None
1071
1072
def __eq__(self, other):
1073
return self.__clause_element__() == other + ' FOO'
1074
1075
class User(Base, fixtures.ComparableEntity):
1076
1077
__tablename__ = 'users'
1078
id = Column('id', Integer, primary_key=True,
1079
test_needs_autoincrement=True)
1080
_name = Column('name', String(50))
1081
name = sa.orm.synonym('_name',
1082
comparator_factory=CustomCompare)
1083
1084
Base.metadata.create_all()
1085
sess = create_session()
1086
u1 = User(name='someuser FOO')
1087
sess.add(u1)
1088
sess.flush()
1089
eq_(sess.query(User).filter(User.name == 'someuser').one(), u1)
1090
1091
def test_synonym_added(self):
1092
1093
class User(Base, fixtures.ComparableEntity):
1094
1095
__tablename__ = 'users'
1096
id = Column('id', Integer, primary_key=True,
1097
test_needs_autoincrement=True)
1098
_name = Column('name', String(50))
1099
1100
def _set_name(self, name):
1101
self._name = 'SOMENAME ' + name
1102
1103
def _get_name(self):
1104
return self._name
1105
1106
name = property(_get_name, _set_name)
1107
1108
User.name = sa.orm.synonym('_name', descriptor=User.name)
1109
Base.metadata.create_all()
1110
sess = create_session()
1111
u1 = User(name='someuser')
1112
eq_(u1.name, 'SOMENAME someuser')
1113
sess.add(u1)
1114
sess.flush()
1115
eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
1116
).one(), u1)
1117
1118
def test_reentrant_compile_via_foreignkey(self):
1119
1120
class User(Base, fixtures.ComparableEntity):
1121
1122
__tablename__ = 'users'
1123
id = Column('id', Integer, primary_key=True,
1124
test_needs_autoincrement=True)
1125
name = Column('name', String(50))
1126
addresses = relationship('Address', backref='user')
1127
1128
class Address(Base, fixtures.ComparableEntity):
1129
1130
__tablename__ = 'addresses'
1131
id = Column('id', Integer, primary_key=True,
1132
test_needs_autoincrement=True)
1133
email = Column('email', String(50))
1134
user_id = Column('user_id', Integer, ForeignKey(User.id))
1135
1136
# previous versions would force a re-entrant mapper compile via
1137
# the User.id inside the ForeignKey but this is no longer the
1138
# case
1139
1140
sa.orm.configure_mappers()
1141
eq_(str(list(Address.user_id.property.columns[0].foreign_keys)[0]),
1142
"ForeignKey('users.id')")
1143
Base.metadata.create_all()
1144
u1 = User(name='u1', addresses=[Address(email='one'),
1145
Address(email='two')])
1146
sess = create_session()
1147
sess.add(u1)
1148
sess.flush()
1149
sess.expunge_all()
1150
eq_(sess.query(User).all(), [User(name='u1',
1151
addresses=[Address(email='one'), Address(email='two')])])
1152
1153
def test_relationship_reference(self):
1154
1155
class Address(Base, fixtures.ComparableEntity):
1156
1157
__tablename__ = 'addresses'
1158
id = Column('id', Integer, primary_key=True,
1159
test_needs_autoincrement=True)
1160
email = Column('email', String(50))
1161
user_id = Column('user_id', Integer, ForeignKey('users.id'))
1162
1163
class User(Base, fixtures.ComparableEntity):
1164
1165
__tablename__ = 'users'
1166
id = Column('id', Integer, primary_key=True,
1167
test_needs_autoincrement=True)
1168
name = Column('name', String(50))
1169
addresses = relationship('Address', backref='user',
1170
primaryjoin=id == Address.user_id)
1171
1172
User.address_count = \
1173
sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
1174
where(Address.user_id
1175
== User.id).as_scalar())
1176
Base.metadata.create_all()
1177
u1 = User(name='u1', addresses=[Address(email='one'),
1178
Address(email='two')])
1179
sess = create_session()
1180
sess.add(u1)
1181
sess.flush()
1182
sess.expunge_all()
1183
eq_(sess.query(User).all(), [User(name='u1', address_count=2,
1184
addresses=[Address(email='one'), Address(email='two')])])
1185
1186
def test_pk_with_fk_init(self):
1187
1188
class Bar(Base):
1189
1190
__tablename__ = 'bar'
1191
id = sa.Column(sa.Integer, sa.ForeignKey('foo.id'),
1192
primary_key=True)
1193
ex = sa.Column(sa.Integer, primary_key=True)
1194
1195
class Foo(Base):
1196
1197
__tablename__ = 'foo'
1198
id = sa.Column(sa.Integer, primary_key=True)
1199
bars = sa.orm.relationship(Bar)
1200
1201
assert Bar.__mapper__.primary_key[0] is Bar.__table__.c.id
1202
assert Bar.__mapper__.primary_key[1] is Bar.__table__.c.ex
1203
1204
def test_with_explicit_autoloaded(self):
1205
meta = MetaData(testing.db)
1206
t1 = Table('t1', meta, Column('id', String(50),
1207
primary_key=True, test_needs_autoincrement=True),
1208
Column('data', String(50)))
1209
meta.create_all()
1210
try:
1211
1212
class MyObj(Base):
1213
1214
__table__ = Table('t1', Base.metadata, autoload=True)
1215
1216
sess = create_session()
1217
m = MyObj(id='someid', data='somedata')
1218
sess.add(m)
1219
sess.flush()
1220
eq_(t1.select().execute().fetchall(), [('someid', 'somedata'
1221
)])
1222
finally:
1223
meta.drop_all()
1224
1225
def test_synonym_for(self):
1226
1227
class User(Base, fixtures.ComparableEntity):
1228
1229
__tablename__ = 'users'
1230
id = Column('id', Integer, primary_key=True,
1231
test_needs_autoincrement=True)
1232
name = Column('name', String(50))
1233
1234
@decl.synonym_for('name')
1235
@property
1236
def namesyn(self):
1237
return self.name
1238
1239
Base.metadata.create_all()
1240
sess = create_session()
1241
u1 = User(name='someuser')
1242
eq_(u1.name, 'someuser')
1243
eq_(u1.namesyn, 'someuser')
1244
sess.add(u1)
1245
sess.flush()
1246
rt = sess.query(User).filter(User.namesyn == 'someuser').one()
1247
eq_(rt, u1)
1248
1249
def test_comparable_using(self):
1250
1251
class NameComparator(sa.orm.PropComparator):
1252
1253
@property
1254
def upperself(self):
1255
cls = self.prop.parent.class_
1256
col = getattr(cls, 'name')
1257
return sa.func.upper(col)
1258
1259
def operate(
1260
self,
1261
op,
1262
other,
1263
**kw
1264
):
1265
return op(self.upperself, other, **kw)
1266
1267
class User(Base, fixtures.ComparableEntity):
1268
1269
__tablename__ = 'users'
1270
id = Column('id', Integer, primary_key=True,
1271
test_needs_autoincrement=True)
1272
name = Column('name', String(50))
1273
1274
@decl.comparable_using(NameComparator)
1275
@property
1276
def uc_name(self):
1277
return self.name is not None and self.name.upper() \
1278
or None
1279
1280
Base.metadata.create_all()
1281
sess = create_session()
1282
u1 = User(name='someuser')
1283
eq_(u1.name, 'someuser', u1.name)
1284
eq_(u1.uc_name, 'SOMEUSER', u1.uc_name)
1285
sess.add(u1)
1286
sess.flush()
1287
sess.expunge_all()
1288
rt = sess.query(User).filter(User.uc_name == 'SOMEUSER').one()
1289
eq_(rt, u1)
1290
sess.expunge_all()
1291
rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE'
1292
)).one()
1293
eq_(rt, u1)
1294
1295
@testing.emits_warning(
1296
"The classname 'Test' is already in the registry "
1297
"of this declarative base, mapped to "
1298
"<class 'test.ext.test_declarative..*Test'>"
1299
)
1300
def test_duplicate_classes_in_base(self):
1301
1302
class Test(Base):
1303
__tablename__ = 'a'
1304
id = Column(Integer, primary_key=True)
1305
1306
class Test(Base):
1307
__tablename__ = 'b'
1308
id = Column(Integer, primary_key=True)
1309
1310
class DeclarativeInheritanceTest(DeclarativeTestBase):
1311
1312
def test_we_must_copy_mapper_args(self):
1313
1314
class Person(Base):
1315
1316
__tablename__ = 'people'
1317
id = Column(Integer, primary_key=True)
1318
discriminator = Column('type', String(50))
1319
__mapper_args__ = {'polymorphic_on': discriminator,
1320
'polymorphic_identity': 'person'}
1321
1322
class Engineer(Person):
1323
1324
primary_language = Column(String(50))
1325
1326
assert 'inherits' not in Person.__mapper_args__
1327
assert class_mapper(Engineer).polymorphic_identity is None
1328
assert class_mapper(Engineer).polymorphic_on is Person.__table__.c.type
1329
1330
def test_we_must_only_copy_column_mapper_args(self):
1331
1332
class Person(Base):
1333
1334
__tablename__ = 'people'
1335
id = Column(Integer, primary_key=True)
1336
a=Column(Integer)
1337
b=Column(Integer)
1338
c=Column(Integer)
1339
d=Column(Integer)
1340
discriminator = Column('type', String(50))
1341
__mapper_args__ = {'polymorphic_on': discriminator,
1342
'polymorphic_identity': 'person',
1343
'version_id_col': 'a',
1344
'column_prefix': 'bar',
1345
'include_properties': ['id', 'a', 'b'],
1346
}
1347
assert class_mapper(Person).version_id_col == 'a'
1348
assert class_mapper(Person).include_properties == set(['id', 'a', 'b'])
1349
1350
1351
def test_custom_join_condition(self):
1352
1353
class Foo(Base):
1354
1355
__tablename__ = 'foo'
1356
id = Column('id', Integer, primary_key=True)
1357
1358
class Bar(Foo):
1359
1360
__tablename__ = 'bar'
1361
id = Column('id', Integer, primary_key=True)
1362
foo_id = Column('foo_id', Integer)
1363
__mapper_args__ = {'inherit_condition': foo_id == Foo.id}
1364
1365
# compile succeeds because inherit_condition is honored
1366
1367
configure_mappers()
1368
1369
def test_joined(self):
1370
1371
class Company(Base, fixtures.ComparableEntity):
1372
1373
__tablename__ = 'companies'
1374
id = Column('id', Integer, primary_key=True,
1375
test_needs_autoincrement=True)
1376
name = Column('name', String(50))
1377
employees = relationship('Person')
1378
1379
class Person(Base, fixtures.ComparableEntity):
1380
1381
__tablename__ = 'people'
1382
id = Column('id', Integer, primary_key=True,
1383
test_needs_autoincrement=True)
1384
company_id = Column('company_id', Integer,
1385
ForeignKey('companies.id'))
1386
name = Column('name', String(50))
1387
discriminator = Column('type', String(50))
1388
__mapper_args__ = {'polymorphic_on': discriminator}
1389
1390
class Engineer(Person):
1391
1392
__tablename__ = 'engineers'
1393
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1394
id = Column('id', Integer, ForeignKey('people.id'),
1395
primary_key=True)
1396
primary_language = Column('primary_language', String(50))
1397
1398
class Manager(Person):
1399
1400
__tablename__ = 'managers'
1401
__mapper_args__ = {'polymorphic_identity': 'manager'}
1402
id = Column('id', Integer, ForeignKey('people.id'),
1403
primary_key=True)
1404
golf_swing = Column('golf_swing', String(50))
1405
1406
Base.metadata.create_all()
1407
sess = create_session()
1408
c1 = Company(name='MegaCorp, Inc.',
1409
employees=[Engineer(name='dilbert',
1410
primary_language='java'), Engineer(name='wally',
1411
primary_language='c++'), Manager(name='dogbert',
1412
golf_swing='fore!')])
1413
c2 = Company(name='Elbonia, Inc.',
1414
employees=[Engineer(name='vlad',
1415
primary_language='cobol')])
1416
sess.add(c1)
1417
sess.add(c2)
1418
sess.flush()
1419
sess.expunge_all()
1420
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
1421
any(Engineer.primary_language
1422
== 'cobol')).first(), c2)
1423
1424
# ensure that the Manager mapper was compiled with the Manager id
1425
# column as higher priority. this ensures that "Manager.id"
1426
# is appropriately treated as the "id" column in the "manager"
1427
# table (reversed from 0.6's behavior.)
1428
1429
assert Manager.id.property.columns == [Manager.__table__.c.id, Person.__table__.c.id]
1430
1431
# assert that the "id" column is available without a second
1432
# load. as of 0.7, the ColumnProperty tests all columns
1433
# in it's list to see which is present in the row.
1434
1435
sess.expunge_all()
1436
1437
def go():
1438
assert sess.query(Manager).filter(Manager.name == 'dogbert'
1439
).one().id
1440
self.assert_sql_count(testing.db, go, 1)
1441
sess.expunge_all()
1442
1443
def go():
1444
assert sess.query(Person).filter(Manager.name == 'dogbert'
1445
).one().id
1446
1447
self.assert_sql_count(testing.db, go, 1)
1448
1449
def test_add_subcol_after_the_fact(self):
1450
1451
class Person(Base, fixtures.ComparableEntity):
1452
1453
__tablename__ = 'people'
1454
id = Column('id', Integer, primary_key=True,
1455
test_needs_autoincrement=True)
1456
name = Column('name', String(50))
1457
discriminator = Column('type', String(50))
1458
__mapper_args__ = {'polymorphic_on': discriminator}
1459
1460
class Engineer(Person):
1461
1462
__tablename__ = 'engineers'
1463
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1464
id = Column('id', Integer, ForeignKey('people.id'),
1465
primary_key=True)
1466
1467
Engineer.primary_language = Column('primary_language',
1468
String(50))
1469
Base.metadata.create_all()
1470
sess = create_session()
1471
e1 = Engineer(primary_language='java', name='dilbert')
1472
sess.add(e1)
1473
sess.flush()
1474
sess.expunge_all()
1475
eq_(sess.query(Person).first(), Engineer(primary_language='java'
1476
, name='dilbert'))
1477
1478
def test_add_parentcol_after_the_fact(self):
1479
1480
class Person(Base, fixtures.ComparableEntity):
1481
1482
__tablename__ = 'people'
1483
id = Column('id', Integer, primary_key=True,
1484
test_needs_autoincrement=True)
1485
discriminator = Column('type', String(50))
1486
__mapper_args__ = {'polymorphic_on': discriminator}
1487
1488
class Engineer(Person):
1489
1490
__tablename__ = 'engineers'
1491
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1492
primary_language = Column(String(50))
1493
id = Column('id', Integer, ForeignKey('people.id'),
1494
primary_key=True)
1495
1496
Person.name = Column('name', String(50))
1497
Base.metadata.create_all()
1498
sess = create_session()
1499
e1 = Engineer(primary_language='java', name='dilbert')
1500
sess.add(e1)
1501
sess.flush()
1502
sess.expunge_all()
1503
eq_(sess.query(Person).first(),
1504
Engineer(primary_language='java', name='dilbert'))
1505
1506
def test_add_sub_parentcol_after_the_fact(self):
1507
1508
class Person(Base, fixtures.ComparableEntity):
1509
1510
__tablename__ = 'people'
1511
id = Column('id', Integer, primary_key=True,
1512
test_needs_autoincrement=True)
1513
discriminator = Column('type', String(50))
1514
__mapper_args__ = {'polymorphic_on': discriminator}
1515
1516
class Engineer(Person):
1517
1518
__tablename__ = 'engineers'
1519
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1520
primary_language = Column(String(50))
1521
id = Column('id', Integer, ForeignKey('people.id'),
1522
primary_key=True)
1523
1524
class Admin(Engineer):
1525
1526
__tablename__ = 'admins'
1527
__mapper_args__ = {'polymorphic_identity': 'admin'}
1528
workstation = Column(String(50))
1529
id = Column('id', Integer, ForeignKey('engineers.id'),
1530
primary_key=True)
1531
1532
Person.name = Column('name', String(50))
1533
Base.metadata.create_all()
1534
sess = create_session()
1535
e1 = Admin(primary_language='java', name='dilbert',
1536
workstation='foo')
1537
sess.add(e1)
1538
sess.flush()
1539
sess.expunge_all()
1540
eq_(sess.query(Person).first(), Admin(primary_language='java',
1541
name='dilbert', workstation='foo'))
1542
1543
def test_subclass_mixin(self):
1544
1545
class Person(Base, fixtures.ComparableEntity):
1546
1547
__tablename__ = 'people'
1548
id = Column('id', Integer, primary_key=True)
1549
name = Column('name', String(50))
1550
discriminator = Column('type', String(50))
1551
__mapper_args__ = {'polymorphic_on': discriminator}
1552
1553
class MyMixin(object):
1554
1555
pass
1556
1557
class Engineer(MyMixin, Person):
1558
1559
__tablename__ = 'engineers'
1560
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1561
id = Column('id', Integer, ForeignKey('people.id'),
1562
primary_key=True)
1563
primary_language = Column('primary_language', String(50))
1564
1565
assert class_mapper(Engineer).inherits is class_mapper(Person)
1566
1567
@testing.fails_if(lambda: True, "Not implemented until 0.7")
1568
def test_foreign_keys_with_col(self):
1569
"""Test that foreign keys that reference a literal 'id' subclass
1570
'id' attribute behave intuitively.
1571
1572
See [ticket:1892].
1573
1574
"""
1575
1576
class Booking(Base):
1577
__tablename__ = 'booking'
1578
id = Column(Integer, primary_key=True)
1579
1580
class PlanBooking(Booking):
1581
__tablename__ = 'plan_booking'
1582
id = Column(Integer, ForeignKey(Booking.id),
1583
primary_key=True)
1584
1585
# referencing PlanBooking.id gives us the column
1586
# on plan_booking, not booking
1587
class FeatureBooking(Booking):
1588
__tablename__ = 'feature_booking'
1589
id = Column(Integer, ForeignKey(Booking.id),
1590
primary_key=True)
1591
plan_booking_id = Column(Integer,
1592
ForeignKey(PlanBooking.id))
1593
1594
plan_booking = relationship(PlanBooking,
1595
backref='feature_bookings')
1596
1597
assert FeatureBooking.__table__.c.plan_booking_id.\
1598
references(PlanBooking.__table__.c.id)
1599
1600
assert FeatureBooking.__table__.c.id.\
1601
references(Booking.__table__.c.id)
1602
1603
def test_with_undefined_foreignkey(self):
1604
1605
class Parent(Base):
1606
1607
__tablename__ = 'parent'
1608
id = Column('id', Integer, primary_key=True)
1609
tp = Column('type', String(50))
1610
__mapper_args__ = dict(polymorphic_on=tp)
1611
1612
class Child1(Parent):
1613
1614
__tablename__ = 'child1'
1615
id = Column('id', Integer, ForeignKey('parent.id'),
1616
primary_key=True)
1617
related_child2 = Column('c2', Integer,
1618
ForeignKey('child2.id'))
1619
__mapper_args__ = dict(polymorphic_identity='child1')
1620
1621
# no exception is raised by the ForeignKey to "child2" even
1622
# though child2 doesn't exist yet
1623
1624
class Child2(Parent):
1625
1626
__tablename__ = 'child2'
1627
id = Column('id', Integer, ForeignKey('parent.id'),
1628
primary_key=True)
1629
related_child1 = Column('c1', Integer)
1630
__mapper_args__ = dict(polymorphic_identity='child2')
1631
1632
sa.orm.configure_mappers() # no exceptions here
1633
1634
def test_foreign_keys_with_col(self):
1635
"""Test that foreign keys that reference a literal 'id' subclass
1636
'id' attribute behave intuitively.
1637
1638
See [ticket:1892].
1639
1640
"""
1641
1642
class Booking(Base):
1643
__tablename__ = 'booking'
1644
id = Column(Integer, primary_key=True)
1645
1646
class PlanBooking(Booking):
1647
__tablename__ = 'plan_booking'
1648
id = Column(Integer, ForeignKey(Booking.id),
1649
primary_key=True)
1650
1651
# referencing PlanBooking.id gives us the column
1652
# on plan_booking, not booking
1653
class FeatureBooking(Booking):
1654
__tablename__ = 'feature_booking'
1655
id = Column(Integer, ForeignKey(Booking.id),
1656
primary_key=True)
1657
plan_booking_id = Column(Integer,
1658
ForeignKey(PlanBooking.id))
1659
1660
plan_booking = relationship(PlanBooking,
1661
backref='feature_bookings')
1662
1663
assert FeatureBooking.__table__.c.plan_booking_id.\
1664
references(PlanBooking.__table__.c.id)
1665
1666
assert FeatureBooking.__table__.c.id.\
1667
references(Booking.__table__.c.id)
1668
1669
1670
def test_single_colsonbase(self):
1671
"""test single inheritance where all the columns are on the base
1672
class."""
1673
1674
class Company(Base, fixtures.ComparableEntity):
1675
1676
__tablename__ = 'companies'
1677
id = Column('id', Integer, primary_key=True,
1678
test_needs_autoincrement=True)
1679
name = Column('name', String(50))
1680
employees = relationship('Person')
1681
1682
class Person(Base, fixtures.ComparableEntity):
1683
1684
__tablename__ = 'people'
1685
id = Column('id', Integer, primary_key=True,
1686
test_needs_autoincrement=True)
1687
company_id = Column('company_id', Integer,
1688
ForeignKey('companies.id'))
1689
name = Column('name', String(50))
1690
discriminator = Column('type', String(50))
1691
primary_language = Column('primary_language', String(50))
1692
golf_swing = Column('golf_swing', String(50))
1693
__mapper_args__ = {'polymorphic_on': discriminator}
1694
1695
class Engineer(Person):
1696
1697
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1698
1699
class Manager(Person):
1700
1701
__mapper_args__ = {'polymorphic_identity': 'manager'}
1702
1703
Base.metadata.create_all()
1704
sess = create_session()
1705
c1 = Company(name='MegaCorp, Inc.',
1706
employees=[Engineer(name='dilbert',
1707
primary_language='java'), Engineer(name='wally',
1708
primary_language='c++'), Manager(name='dogbert',
1709
golf_swing='fore!')])
1710
c2 = Company(name='Elbonia, Inc.',
1711
employees=[Engineer(name='vlad',
1712
primary_language='cobol')])
1713
sess.add(c1)
1714
sess.add(c2)
1715
sess.flush()
1716
sess.expunge_all()
1717
eq_(sess.query(Person).filter(Engineer.primary_language
1718
== 'cobol').first(), Engineer(name='vlad'))
1719
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
1720
any(Engineer.primary_language
1721
== 'cobol')).first(), c2)
1722
1723
def test_single_colsonsub(self):
1724
"""test single inheritance where the columns are local to their
1725
class.
1726
1727
this is a newer usage.
1728
1729
"""
1730
1731
class Company(Base, fixtures.ComparableEntity):
1732
1733
__tablename__ = 'companies'
1734
id = Column('id', Integer, primary_key=True,
1735
test_needs_autoincrement=True)
1736
name = Column('name', String(50))
1737
employees = relationship('Person')
1738
1739
class Person(Base, fixtures.ComparableEntity):
1740
1741
__tablename__ = 'people'
1742
id = Column(Integer, primary_key=True,
1743
test_needs_autoincrement=True)
1744
company_id = Column(Integer, ForeignKey('companies.id'))
1745
name = Column(String(50))
1746
discriminator = Column('type', String(50))
1747
__mapper_args__ = {'polymorphic_on': discriminator}
1748
1749
class Engineer(Person):
1750
1751
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1752
primary_language = Column(String(50))
1753
1754
class Manager(Person):
1755
1756
__mapper_args__ = {'polymorphic_identity': 'manager'}
1757
golf_swing = Column(String(50))
1758
1759
# we have here a situation that is somewhat unique. the Person
1760
# class is mapped to the "people" table, but it was mapped when
1761
# the table did not include the "primary_language" or
1762
# "golf_swing" columns. declarative will also manipulate the
1763
# exclude_properties collection so that sibling classes don't
1764
# cross-pollinate.
1765
1766
assert Person.__table__.c.company_id is not None
1767
assert Person.__table__.c.golf_swing is not None
1768
assert Person.__table__.c.primary_language is not None
1769
assert Engineer.primary_language is not None
1770
assert Manager.golf_swing is not None
1771
assert not hasattr(Person, 'primary_language')
1772
assert not hasattr(Person, 'golf_swing')
1773
assert not hasattr(Engineer, 'golf_swing')
1774
assert not hasattr(Manager, 'primary_language')
1775
Base.metadata.create_all()
1776
sess = create_session()
1777
e1 = Engineer(name='dilbert', primary_language='java')
1778
e2 = Engineer(name='wally', primary_language='c++')
1779
m1 = Manager(name='dogbert', golf_swing='fore!')
1780
c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1])
1781
e3 = Engineer(name='vlad', primary_language='cobol')
1782
c2 = Company(name='Elbonia, Inc.', employees=[e3])
1783
sess.add(c1)
1784
sess.add(c2)
1785
sess.flush()
1786
sess.expunge_all()
1787
eq_(sess.query(Person).filter(Engineer.primary_language
1788
== 'cobol').first(), Engineer(name='vlad'))
1789
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
1790
any(Engineer.primary_language
1791
== 'cobol')).first(), c2)
1792
eq_(sess.query(Engineer).filter_by(primary_language='cobol'
1793
).one(), Engineer(name='vlad', primary_language='cobol'))
1794
1795
def test_joined_from_single(self):
1796
1797
class Company(Base, fixtures.ComparableEntity):
1798
1799
__tablename__ = 'companies'
1800
id = Column('id', Integer, primary_key=True,
1801
test_needs_autoincrement=True)
1802
name = Column('name', String(50))
1803
employees = relationship('Person')
1804
1805
class Person(Base, fixtures.ComparableEntity):
1806
1807
__tablename__ = 'people'
1808
id = Column(Integer, primary_key=True,
1809
test_needs_autoincrement=True)
1810
company_id = Column(Integer, ForeignKey('companies.id'))
1811
name = Column(String(50))
1812
discriminator = Column('type', String(50))
1813
__mapper_args__ = {'polymorphic_on': discriminator}
1814
1815
class Manager(Person):
1816
1817
__mapper_args__ = {'polymorphic_identity': 'manager'}
1818
golf_swing = Column(String(50))
1819
1820
class Engineer(Person):
1821
1822
__tablename__ = 'engineers'
1823
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1824
id = Column(Integer, ForeignKey('people.id'),
1825
primary_key=True)
1826
primary_language = Column(String(50))
1827
1828
assert Person.__table__.c.golf_swing is not None
1829
assert not Person.__table__.c.has_key('primary_language')
1830
assert Engineer.__table__.c.primary_language is not None
1831
assert Engineer.primary_language is not None
1832
assert Manager.golf_swing is not None
1833
assert not hasattr(Person, 'primary_language')
1834
assert not hasattr(Person, 'golf_swing')
1835
assert not hasattr(Engineer, 'golf_swing')
1836
assert not hasattr(Manager, 'primary_language')
1837
Base.metadata.create_all()
1838
sess = create_session()
1839
e1 = Engineer(name='dilbert', primary_language='java')
1840
e2 = Engineer(name='wally', primary_language='c++')
1841
m1 = Manager(name='dogbert', golf_swing='fore!')
1842
c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1])
1843
e3 = Engineer(name='vlad', primary_language='cobol')
1844
c2 = Company(name='Elbonia, Inc.', employees=[e3])
1845
sess.add(c1)
1846
sess.add(c2)
1847
sess.flush()
1848
sess.expunge_all()
1849
eq_(sess.query(Person).with_polymorphic(Engineer).
1850
filter(Engineer.primary_language
1851
== 'cobol').first(), Engineer(name='vlad'))
1852
eq_(sess.query(Company).filter(Company.employees.of_type(Engineer).
1853
any(Engineer.primary_language
1854
== 'cobol')).first(), c2)
1855
eq_(sess.query(Engineer).filter_by(primary_language='cobol'
1856
).one(), Engineer(name='vlad', primary_language='cobol'))
1857
1858
def test_polymorphic_on_converted_from_inst(self):
1859
class A(Base):
1860
__tablename__ = 'A'
1861
id = Column(Integer, primary_key=True)
1862
discriminator = Column(String)
1863
1864
@declared_attr
1865
def __mapper_args__(cls):
1866
return {
1867
'polymorphic_identity': cls.__name__,
1868
'polymorphic_on': cls.discriminator
1869
}
1870
1871
class B(A):
1872
pass
1873
is_(B.__mapper__.polymorphic_on, A.__table__.c.discriminator)
1874
1875
def test_add_deferred(self):
1876
1877
class Person(Base, fixtures.ComparableEntity):
1878
1879
__tablename__ = 'people'
1880
id = Column('id', Integer, primary_key=True,
1881
test_needs_autoincrement=True)
1882
1883
Person.name = deferred(Column(String(10)))
1884
Base.metadata.create_all()
1885
sess = create_session()
1886
p = Person(name='ratbert')
1887
sess.add(p)
1888
sess.flush()
1889
sess.expunge_all()
1890
eq_(sess.query(Person).all(), [Person(name='ratbert')])
1891
sess.expunge_all()
1892
person = sess.query(Person).filter(Person.name == 'ratbert'
1893
).one()
1894
assert 'name' not in person.__dict__
1895
1896
def test_single_fksonsub(self):
1897
"""test single inheritance with a foreign key-holding column on
1898
a subclass.
1899
1900
"""
1901
1902
class Person(Base, fixtures.ComparableEntity):
1903
1904
__tablename__ = 'people'
1905
id = Column(Integer, primary_key=True,
1906
test_needs_autoincrement=True)
1907
name = Column(String(50))
1908
discriminator = Column('type', String(50))
1909
__mapper_args__ = {'polymorphic_on': discriminator}
1910
1911
class Engineer(Person):
1912
1913
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1914
primary_language_id = Column(Integer,
1915
ForeignKey('languages.id'))
1916
primary_language = relationship('Language')
1917
1918
class Language(Base, fixtures.ComparableEntity):
1919
1920
__tablename__ = 'languages'
1921
id = Column(Integer, primary_key=True,
1922
test_needs_autoincrement=True)
1923
name = Column(String(50))
1924
1925
assert not hasattr(Person, 'primary_language_id')
1926
Base.metadata.create_all()
1927
sess = create_session()
1928
java, cpp, cobol = Language(name='java'), Language(name='cpp'), \
1929
Language(name='cobol')
1930
e1 = Engineer(name='dilbert', primary_language=java)
1931
e2 = Engineer(name='wally', primary_language=cpp)
1932
e3 = Engineer(name='vlad', primary_language=cobol)
1933
sess.add_all([e1, e2, e3])
1934
sess.flush()
1935
sess.expunge_all()
1936
eq_(sess.query(Person).filter(Engineer.primary_language.has(
1937
Language.name
1938
== 'cobol')).first(), Engineer(name='vlad',
1939
primary_language=Language(name='cobol')))
1940
eq_(sess.query(Engineer).filter(Engineer.primary_language.has(
1941
Language.name
1942
== 'cobol')).one(), Engineer(name='vlad',
1943
primary_language=Language(name='cobol')))
1944
eq_(sess.query(Person).join(Engineer.primary_language).order_by(
1945
Language.name).all(),
1946
[Engineer(name='vlad',
1947
primary_language=Language(name='cobol')),
1948
Engineer(name='wally', primary_language=Language(name='cpp'
1949
)), Engineer(name='dilbert',
1950
primary_language=Language(name='java'))])
1951
1952
def test_single_three_levels(self):
1953
1954
class Person(Base, fixtures.ComparableEntity):
1955
1956
__tablename__ = 'people'
1957
id = Column(Integer, primary_key=True)
1958
name = Column(String(50))
1959
discriminator = Column('type', String(50))
1960
__mapper_args__ = {'polymorphic_on': discriminator}
1961
1962
class Engineer(Person):
1963
1964
__mapper_args__ = {'polymorphic_identity': 'engineer'}
1965
primary_language = Column(String(50))
1966
1967
class JuniorEngineer(Engineer):
1968
1969
__mapper_args__ = \
1970
{'polymorphic_identity': 'junior_engineer'}
1971
nerf_gun = Column(String(50))
1972
1973
class Manager(Person):
1974
1975
__mapper_args__ = {'polymorphic_identity': 'manager'}
1976
golf_swing = Column(String(50))
1977
1978
assert JuniorEngineer.nerf_gun
1979
assert JuniorEngineer.primary_language
1980
assert JuniorEngineer.name
1981
assert Manager.golf_swing
1982
assert Engineer.primary_language
1983
assert not hasattr(Engineer, 'golf_swing')
1984
assert not hasattr(Engineer, 'nerf_gun')
1985
assert not hasattr(Manager, 'nerf_gun')
1986
assert not hasattr(Manager, 'primary_language')
1987
1988
def test_single_detects_conflict(self):
1989
1990
class Person(Base):
1991
1992
__tablename__ = 'people'
1993
id = Column(Integer, primary_key=True)
1994
name = Column(String(50))
1995
discriminator = Column('type', String(50))
1996
__mapper_args__ = {'polymorphic_on': discriminator}
1997
1998
class Engineer(Person):
1999
2000
__mapper_args__ = {'polymorphic_identity': 'engineer'}
2001
primary_language = Column(String(50))
2002
2003
# test sibling col conflict
2004
2005
def go():
2006
2007
class Manager(Person):
2008
2009
__mapper_args__ = {'polymorphic_identity': 'manager'}
2010
golf_swing = Column(String(50))
2011
primary_language = Column(String(50))
2012
2013
assert_raises(sa.exc.ArgumentError, go)
2014
2015
# test parent col conflict
2016
2017
def go():
2018
2019
class Salesman(Person):
2020
2021
__mapper_args__ = {'polymorphic_identity': 'manager'}
2022
name = Column(String(50))
2023
2024
assert_raises(sa.exc.ArgumentError, go)
2025
2026
def test_single_no_special_cols(self):
2027
2028
class Person(Base, fixtures.ComparableEntity):
2029
2030
__tablename__ = 'people'
2031
id = Column('id', Integer, primary_key=True)
2032
name = Column('name', String(50))
2033
discriminator = Column('type', String(50))
2034
__mapper_args__ = {'polymorphic_on': discriminator}
2035
2036
def go():
2037
2038
class Engineer(Person):
2039
2040
__mapper_args__ = {'polymorphic_identity': 'engineer'}
2041
primary_language = Column('primary_language',
2042
String(50))
2043
foo_bar = Column(Integer, primary_key=True)
2044
2045
assert_raises_message(sa.exc.ArgumentError, 'place primary key'
2046
, go)
2047
2048
def test_single_no_table_args(self):
2049
2050
class Person(Base, fixtures.ComparableEntity):
2051
2052
__tablename__ = 'people'
2053
id = Column('id', Integer, primary_key=True)
2054
name = Column('name', String(50))
2055
discriminator = Column('type', String(50))
2056
__mapper_args__ = {'polymorphic_on': discriminator}
2057
2058
def go():
2059
2060
class Engineer(Person):
2061
2062
__mapper_args__ = {'polymorphic_identity': 'engineer'}
2063
primary_language = Column('primary_language',
2064
String(50))
2065
2066
# this should be on the Person class, as this is single
2067
# table inheritance, which is why we test that this
2068
# throws an exception!
2069
2070
__table_args__ = {'mysql_engine': 'InnoDB'}
2071
2072
assert_raises_message(sa.exc.ArgumentError,
2073
'place __table_args__', go)
2074
2075
@testing.emits_warning("The classname")
2076
def test_dupe_name_in_hierarchy(self):
2077
class A(Base):
2078
__tablename__ = "a"
2079
id = Column( Integer, primary_key=True)
2080
a_1 = A
2081
class A(a_1):
2082
__tablename__ = 'b'
2083
2084
id = Column(Integer(),ForeignKey(a_1.id), primary_key = True)
2085
2086
assert A.__mapper__.inherits is a_1.__mapper__
2087
2088
class OverlapColPrecedenceTest(DeclarativeTestBase):
2089
"""test #1892 cases when declarative does column precedence."""
2090
2091
def _run_test(self, Engineer, e_id, p_id):
2092
p_table = Base.metadata.tables['person']
2093
e_table = Base.metadata.tables['engineer']
2094
assert Engineer.id.property.columns[0] is e_table.c[e_id]
2095
assert Engineer.id.property.columns[1] is p_table.c[p_id]
2096
2097
def test_basic(self):
2098
class Person(Base):
2099
__tablename__ = 'person'
2100
id = Column(Integer, primary_key=True)
2101
2102
class Engineer(Person):
2103
__tablename__ = 'engineer'
2104
id = Column(Integer, ForeignKey('person.id'), primary_key=True)
2105
2106
self._run_test(Engineer, "id", "id")
2107
2108
def test_alt_name_base(self):
2109
class Person(Base):
2110
__tablename__ = 'person'
2111
id = Column("pid", Integer, primary_key=True)
2112
2113
class Engineer(Person):
2114
__tablename__ = 'engineer'
2115
id = Column(Integer, ForeignKey('person.pid'), primary_key=True)
2116
2117
self._run_test(Engineer, "id", "pid")
2118
2119
def test_alt_name_sub(self):
2120
class Person(Base):
2121
__tablename__ = 'person'
2122
id = Column(Integer, primary_key=True)
2123
2124
class Engineer(Person):
2125
__tablename__ = 'engineer'
2126
id = Column("eid", Integer, ForeignKey('person.id'), primary_key=True)
2127
2128
self._run_test(Engineer, "eid", "id")
2129
2130
def test_alt_name_both(self):
2131
class Person(Base):
2132
__tablename__ = 'person'
2133
id = Column("pid", Integer, primary_key=True)
2134
2135
class Engineer(Person):
2136
__tablename__ = 'engineer'
2137
id = Column("eid", Integer, ForeignKey('person.pid'), primary_key=True)
2138
2139
self._run_test(Engineer, "eid", "pid")
2140
2141
2142
2143
from test.orm.test_events import _RemoveListeners
2144
class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase):
2145
def _roundtrip(self, Employee, Manager, Engineer, Boss,
2146
polymorphic=True, explicit_type=False):
2147
Base.metadata.create_all()
2148
sess = create_session()
2149
e1 = Engineer(name='dilbert', primary_language='java')
2150
e2 = Engineer(name='wally', primary_language='c++')
2151
m1 = Manager(name='dogbert', golf_swing='fore!')
2152
e3 = Engineer(name='vlad', primary_language='cobol')
2153
b1 = Boss(name="pointy haired")
2154
if polymorphic:
2155
for obj in [e1, e2, m1, e3, b1]:
2156
if explicit_type:
2157
eq_(obj.type, obj.__mapper__.polymorphic_identity)
2158
else:
2159
assert_raises_message(
2160
AttributeError,
2161
"does not implement attribute .?'type' "
2162
"at the instance level.",
2163
getattr, obj, "type"
2164
)
2165
else:
2166
assert "type" not in Engineer.__dict__
2167
assert "type" not in Manager.__dict__
2168
assert "type" not in Boss.__dict__
2169
2170
sess.add_all([e1, e2, m1, e3, b1])
2171
sess.flush()
2172
sess.expunge_all()
2173
if polymorphic:
2174
eq_(sess.query(Employee).order_by(Employee.name).all(),
2175
[Engineer(name='dilbert'), Manager(name='dogbert'),
2176
Boss(name='pointy haired'), Engineer(name='vlad'), Engineer(name='wally')])
2177
else:
2178
eq_(sess.query(Engineer).order_by(Engineer.name).all(),
2179
[Engineer(name='dilbert'), Engineer(name='vlad'),
2180
Engineer(name='wally')])
2181
eq_(sess.query(Manager).all(), [Manager(name='dogbert')])
2182
eq_(sess.query(Boss).all(), [Boss(name='pointy haired')])
2183
2184
2185
def test_explicit(self):
2186
engineers = Table('engineers', Base.metadata, Column('id',
2187
Integer, primary_key=True,
2188
test_needs_autoincrement=True), Column('name'
2189
, String(50)), Column('primary_language',
2190
String(50)))
2191
managers = Table('managers', Base.metadata,
2192
Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
2193
Column('name', String(50)),
2194
Column('golf_swing', String(50))
2195
)
2196
boss = Table('boss', Base.metadata,
2197
Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
2198
Column('name', String(50)),
2199
Column('golf_swing', String(50))
2200
)
2201
punion = polymorphic_union({
2202
'engineer': engineers,
2203
'manager' : managers,
2204
'boss': boss}, 'type', 'punion')
2205
2206
class Employee(Base, fixtures.ComparableEntity):
2207
2208
__table__ = punion
2209
__mapper_args__ = {'polymorphic_on': punion.c.type}
2210
2211
class Engineer(Employee):
2212
2213
__table__ = engineers
2214
__mapper_args__ = {'polymorphic_identity': 'engineer',
2215
'concrete': True}
2216
2217
class Manager(Employee):
2218
2219
__table__ = managers
2220
__mapper_args__ = {'polymorphic_identity': 'manager',
2221
'concrete': True}
2222
2223
class Boss(Manager):
2224
__table__ = boss
2225
__mapper_args__ = {'polymorphic_identity': 'boss',
2226
'concrete': True}
2227
2228
self._roundtrip(Employee, Manager, Engineer, Boss)
2229
2230
def test_concrete_inline_non_polymorphic(self):
2231
"""test the example from the declarative docs."""
2232
2233
class Employee(Base, fixtures.ComparableEntity):
2234
2235
__tablename__ = 'people'
2236
id = Column(Integer, primary_key=True,
2237
test_needs_autoincrement=True)
2238
name = Column(String(50))
2239
2240
class Engineer(Employee):
2241
2242
__tablename__ = 'engineers'
2243
__mapper_args__ = {'concrete': True}
2244
id = Column(Integer, primary_key=True,
2245
test_needs_autoincrement=True)
2246
primary_language = Column(String(50))
2247
name = Column(String(50))
2248
2249
class Manager(Employee):
2250
2251
__tablename__ = 'manager'
2252
__mapper_args__ = {'concrete': True}
2253
id = Column(Integer, primary_key=True,
2254
test_needs_autoincrement=True)
2255
golf_swing = Column(String(50))
2256
name = Column(String(50))
2257
2258
class Boss(Manager):
2259
__tablename__ = 'boss'
2260
__mapper_args__ = {'concrete': True}
2261
id = Column(Integer, primary_key=True,
2262
test_needs_autoincrement=True)
2263
golf_swing = Column(String(50))
2264
name = Column(String(50))
2265
2266
self._roundtrip(Employee, Manager, Engineer, Boss, polymorphic=False)
2267
2268
def test_abstract_concrete_extension(self):
2269
class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
2270
pass
2271
2272
class Manager(Employee):
2273
__tablename__ = 'manager'
2274
employee_id = Column(Integer, primary_key=True,
2275
test_needs_autoincrement=True)
2276
name = Column(String(50))
2277
golf_swing = Column(String(40))
2278
__mapper_args__ = {
2279
'polymorphic_identity':'manager',
2280
'concrete':True}
2281
2282
class Boss(Manager):
2283
__tablename__ = 'boss'
2284
employee_id = Column(Integer, primary_key=True,
2285
test_needs_autoincrement=True)
2286
name = Column(String(50))
2287
golf_swing = Column(String(40))
2288
__mapper_args__ = {
2289
'polymorphic_identity':'boss',
2290
'concrete':True}
2291
2292
class Engineer(Employee):
2293
__tablename__ = 'engineer'
2294
employee_id = Column(Integer, primary_key=True,
2295
test_needs_autoincrement=True)
2296
name = Column(String(50))
2297
primary_language = Column(String(40))
2298
__mapper_args__ = {'polymorphic_identity':'engineer',
2299
'concrete':True}
2300
2301
self._roundtrip(Employee, Manager, Engineer, Boss)
2302
2303
def test_concrete_extension(self):
2304
class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
2305
__tablename__ = 'employee'
2306
employee_id = Column(Integer, primary_key=True,
2307
test_needs_autoincrement=True)
2308
name = Column(String(50))
2309
__mapper_args__ = {
2310
'polymorphic_identity':'employee',
2311
'concrete':True}
2312
class Manager(Employee):
2313
__tablename__ = 'manager'
2314
employee_id = Column(Integer, primary_key=True,
2315
test_needs_autoincrement=True)
2316
name = Column(String(50))
2317
golf_swing = Column(String(40))
2318
__mapper_args__ = {
2319
'polymorphic_identity':'manager',
2320
'concrete':True}
2321
2322
class Boss(Manager):
2323
__tablename__ = 'boss'
2324
employee_id = Column(Integer, primary_key=True,
2325
test_needs_autoincrement=True)
2326
name = Column(String(50))
2327
golf_swing = Column(String(40))
2328
__mapper_args__ = {
2329
'polymorphic_identity':'boss',
2330
'concrete':True}
2331
2332
class Engineer(Employee):
2333
__tablename__ = 'engineer'
2334
employee_id = Column(Integer, primary_key=True,
2335
test_needs_autoincrement=True)
2336
name = Column(String(50))
2337
primary_language = Column(String(40))
2338
__mapper_args__ = {'polymorphic_identity':'engineer',
2339
'concrete':True}
2340
self._roundtrip(Employee, Manager, Engineer, Boss)
2341
2342
2343
def test_ok_to_override_type_from_abstract(self):
2344
class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
2345
pass
2346
2347
class Manager(Employee):
2348
__tablename__ = 'manager'
2349
employee_id = Column(Integer, primary_key=True,
2350
test_needs_autoincrement=True)
2351
name = Column(String(50))
2352
golf_swing = Column(String(40))
2353
2354
@property
2355
def type(self):
2356
return "manager"
2357
2358
__mapper_args__ = {
2359
'polymorphic_identity': "manager",
2360
'concrete':True}
2361
2362
class Boss(Manager):
2363
__tablename__ = 'boss'
2364
employee_id = Column(Integer, primary_key=True,
2365
test_needs_autoincrement=True)
2366
name = Column(String(50))
2367
golf_swing = Column(String(40))
2368
2369
@property
2370
def type(self):
2371
return "boss"
2372
2373
__mapper_args__ = {
2374
'polymorphic_identity': "boss",
2375
'concrete':True}
2376
2377
class Engineer(Employee):
2378
__tablename__ = 'engineer'
2379
employee_id = Column(Integer, primary_key=True,
2380
test_needs_autoincrement=True)
2381
name = Column(String(50))
2382
primary_language = Column(String(40))
2383
2384
@property
2385
def type(self):
2386
return "engineer"
2387
__mapper_args__ = {'polymorphic_identity': "engineer",
2388
'concrete':True}
2389
self._roundtrip(Employee, Manager, Engineer, Boss, explicit_type=True)
2390
2391
def _produce_test(inline, stringbased):
2392
2393
class ExplicitJoinTest(fixtures.MappedTest):
2394
2395
@classmethod
2396
def define_tables(cls, metadata):
2397
global User, Address
2398
Base = decl.declarative_base(metadata=metadata)
2399
2400
class User(Base, fixtures.ComparableEntity):
2401
2402
__tablename__ = 'users'
2403
id = Column(Integer, primary_key=True,
2404
test_needs_autoincrement=True)
2405
name = Column(String(50))
2406
2407
class Address(Base, fixtures.ComparableEntity):
2408
2409
__tablename__ = 'addresses'
2410
id = Column(Integer, primary_key=True,
2411
test_needs_autoincrement=True)
2412
email = Column(String(50))
2413
user_id = Column(Integer, ForeignKey('users.id'))
2414
if inline:
2415
if stringbased:
2416
user = relationship('User',
2417
primaryjoin='User.id==Address.user_id',
2418
backref='addresses')
2419
else:
2420
user = relationship(User, primaryjoin=User.id
2421
== user_id, backref='addresses')
2422
2423
if not inline:
2424
configure_mappers()
2425
if stringbased:
2426
Address.user = relationship('User',
2427
primaryjoin='User.id==Address.user_id',
2428
backref='addresses')
2429
else:
2430
Address.user = relationship(User,
2431
primaryjoin=User.id == Address.user_id,
2432
backref='addresses')
2433
2434
@classmethod
2435
def insert_data(cls):
2436
params = [dict(zip(('id', 'name'), column_values))
2437
for column_values in [(7, 'jack'), (8, 'ed'), (9,
2438
'fred'), (10, 'chuck')]]
2439
User.__table__.insert().execute(params)
2440
Address.__table__.insert().execute([dict(zip(('id',
2441
'user_id', 'email'), column_values))
2442
for column_values in [(1, 7, 'jack@bean.com'), (2,
2443
8, 'ed@wood.com'), (3, 8, 'ed@bettyboop.com'), (4,
2444
8, 'ed@lala.com'), (5, 9, 'fred@fred.com')]])
2445
2446
def test_aliased_join(self):
2447
2448
# this query will screw up if the aliasing enabled in
2449
# query.join() gets applied to the right half of the join
2450
# condition inside the any(). the join condition inside of
2451
# any() comes from the "primaryjoin" of the relationship,
2452
# and should not be annotated with _orm_adapt.
2453
# PropertyLoader.Comparator will annotate the left side with
2454
# _orm_adapt, though.
2455
2456
sess = create_session()
2457
eq_(sess.query(User).join(User.addresses,
2458
aliased=True).filter(Address.email == 'ed@wood.com'
2459
).filter(User.addresses.any(Address.email
2460
== 'jack@bean.com')).all(), [])
2461
2462
ExplicitJoinTest.__name__ = 'ExplicitJoinTest%s%s' % (inline
2463
and 'Inline' or 'Separate', stringbased and 'String'
2464
or 'Literal')
2465
return ExplicitJoinTest
2466
2467
for inline in True, False:
2468
for stringbased in True, False:
2469
testclass = _produce_test(inline, stringbased)
2470
exec '%s = testclass' % testclass.__name__
2471
del testclass
2472
2473
class DeclarativeReflectionTest(fixtures.TestBase):
2474
2475
@classmethod
2476
def setup_class(cls):
2477
global reflection_metadata
2478
reflection_metadata = MetaData(testing.db)
2479
Table('users', reflection_metadata, Column('id', Integer,
2480
primary_key=True, test_needs_autoincrement=True),
2481
Column('name', String(50)), test_needs_fk=True)
2482
Table(
2483
'addresses',
2484
reflection_metadata,
2485
Column('id', Integer, primary_key=True,
2486
test_needs_autoincrement=True),
2487
Column('email', String(50)),
2488
Column('user_id', Integer, ForeignKey('users.id')),
2489
test_needs_fk=True,
2490
)
2491
Table(
2492
'imhandles',
2493
reflection_metadata,
2494
Column('id', Integer, primary_key=True,
2495
test_needs_autoincrement=True),
2496
Column('user_id', Integer),
2497
Column('network', String(50)),
2498
Column('handle', String(50)),
2499
test_needs_fk=True,
2500
)
2501
reflection_metadata.create_all()
2502
2503
def setup(self):
2504
global Base
2505
Base = decl.declarative_base(testing.db)
2506
2507
def teardown(self):
2508
for t in reversed(reflection_metadata.sorted_tables):
2509
t.delete().execute()
2510
2511
@classmethod
2512
def teardown_class(cls):
2513
reflection_metadata.drop_all()
2514
2515
def test_basic(self):
2516
meta = MetaData(testing.db)
2517
2518
class User(Base, fixtures.ComparableEntity):
2519
2520
__tablename__ = 'users'
2521
__autoload__ = True
2522
if testing.against('oracle', 'firebird'):
2523
id = Column('id', Integer, primary_key=True,
2524
test_needs_autoincrement=True)
2525
addresses = relationship('Address', backref='user')
2526
2527
class Address(Base, fixtures.ComparableEntity):
2528
2529
__tablename__ = 'addresses'
2530
__autoload__ = True
2531
if testing.against('oracle', 'firebird'):
2532
id = Column('id', Integer, primary_key=True,
2533
test_needs_autoincrement=True)
2534
2535
u1 = User(name='u1', addresses=[Address(email='one'),
2536
Address(email='two')])
2537
sess = create_session()
2538
sess.add(u1)
2539
sess.flush()
2540
sess.expunge_all()
2541
eq_(sess.query(User).all(), [User(name='u1',
2542
addresses=[Address(email='one'), Address(email='two')])])
2543
a1 = sess.query(Address).filter(Address.email == 'two').one()
2544
eq_(a1, Address(email='two'))
2545
eq_(a1.user, User(name='u1'))
2546
2547
def test_rekey(self):
2548
meta = MetaData(testing.db)
2549
2550
class User(Base, fixtures.ComparableEntity):
2551
2552
__tablename__ = 'users'
2553
__autoload__ = True
2554
if testing.against('oracle', 'firebird'):
2555
id = Column('id', Integer, primary_key=True,
2556
test_needs_autoincrement=True)
2557
nom = Column('name', String(50), key='nom')
2558
addresses = relationship('Address', backref='user')
2559
2560
class Address(Base, fixtures.ComparableEntity):
2561
2562
__tablename__ = 'addresses'
2563
__autoload__ = True
2564
if testing.against('oracle', 'firebird'):
2565
id = Column('id', Integer, primary_key=True,
2566
test_needs_autoincrement=True)
2567
2568
u1 = User(nom='u1', addresses=[Address(email='one'),
2569
Address(email='two')])
2570
sess = create_session()
2571
sess.add(u1)
2572
sess.flush()
2573
sess.expunge_all()
2574
eq_(sess.query(User).all(), [User(nom='u1',
2575
addresses=[Address(email='one'), Address(email='two')])])
2576
a1 = sess.query(Address).filter(Address.email == 'two').one()
2577
eq_(a1, Address(email='two'))
2578
eq_(a1.user, User(nom='u1'))
2579
assert_raises(TypeError, User, name='u3')
2580
2581
def test_supplied_fk(self):
2582
meta = MetaData(testing.db)
2583
2584
class IMHandle(Base, fixtures.ComparableEntity):
2585
2586
__tablename__ = 'imhandles'
2587
__autoload__ = True
2588
if testing.against('oracle', 'firebird'):
2589
id = Column('id', Integer, primary_key=True,
2590
test_needs_autoincrement=True)
2591
user_id = Column('user_id', Integer, ForeignKey('users.id'))
2592
2593
class User(Base, fixtures.ComparableEntity):
2594
2595
__tablename__ = 'users'
2596
__autoload__ = True
2597
if testing.against('oracle', 'firebird'):
2598
id = Column('id', Integer, primary_key=True,
2599
test_needs_autoincrement=True)
2600
handles = relationship('IMHandle', backref='user')
2601
2602
u1 = User(name='u1', handles=[IMHandle(network='blabber',
2603
handle='foo'), IMHandle(network='lol', handle='zomg'
2604
)])
2605
sess = create_session()
2606
sess.add(u1)
2607
sess.flush()
2608
sess.expunge_all()
2609
eq_(sess.query(User).all(), [User(name='u1',
2610
handles=[IMHandle(network='blabber', handle='foo'),
2611
IMHandle(network='lol', handle='zomg')])])
2612
a1 = sess.query(IMHandle).filter(IMHandle.handle == 'zomg'
2613
).one()
2614
eq_(a1, IMHandle(network='lol', handle='zomg'))
2615
eq_(a1.user, User(name='u1'))
2616
2617
class DeclarativeMixinTest(DeclarativeTestBase):
2618
2619
def test_simple(self):
2620
2621
class MyMixin(object):
2622
2623
id = Column(Integer, primary_key=True,
2624
test_needs_autoincrement=True)
2625
2626
def foo(self):
2627
return 'bar' + str(self.id)
2628
2629
class MyModel(Base, MyMixin):
2630
2631
__tablename__ = 'test'
2632
name = Column(String(100), nullable=False, index=True)
2633
2634
Base.metadata.create_all()
2635
session = create_session()
2636
session.add(MyModel(name='testing'))
2637
session.flush()
2638
session.expunge_all()
2639
obj = session.query(MyModel).one()
2640
eq_(obj.id, 1)
2641
eq_(obj.name, 'testing')
2642
eq_(obj.foo(), 'bar1')
2643
2644
def test_unique_column(self):
2645
2646
class MyMixin(object):
2647
2648
id = Column(Integer, primary_key=True)
2649
value = Column(String, unique=True)
2650
2651
class MyModel(Base, MyMixin):
2652
2653
__tablename__ = 'test'
2654
2655
assert MyModel.__table__.c.value.unique
2656
2657
def test_hierarchical_bases(self):
2658
2659
class MyMixinParent:
2660
2661
id = Column(Integer, primary_key=True,
2662
test_needs_autoincrement=True)
2663
2664
def foo(self):
2665
return 'bar' + str(self.id)
2666
2667
class MyMixin(MyMixinParent):
2668
2669
baz = Column(String(100), nullable=False, index=True)
2670
2671
class MyModel(Base, MyMixin):
2672
2673
__tablename__ = 'test'
2674
name = Column(String(100), nullable=False, index=True)
2675
2676
Base.metadata.create_all()
2677
session = create_session()
2678
session.add(MyModel(name='testing', baz='fu'))
2679
session.flush()
2680
session.expunge_all()
2681
obj = session.query(MyModel).one()
2682
eq_(obj.id, 1)
2683
eq_(obj.name, 'testing')
2684
eq_(obj.foo(), 'bar1')
2685
eq_(obj.baz, 'fu')
2686
2687
def test_mixin_overrides(self):
2688
"""test a mixin that overrides a column on a superclass."""
2689
2690
class MixinA(object):
2691
foo = Column(String(50))
2692
2693
class MixinB(MixinA):
2694
foo = Column(Integer)
2695
2696
class MyModelA(Base, MixinA):
2697
__tablename__ = 'testa'
2698
id = Column(Integer, primary_key=True)
2699
2700
class MyModelB(Base, MixinB):
2701
__tablename__ = 'testb'
2702
id = Column(Integer, primary_key=True)
2703
2704
eq_(MyModelA.__table__.c.foo.type.__class__, String)
2705
eq_(MyModelB.__table__.c.foo.type.__class__, Integer)
2706
2707
2708
def test_not_allowed(self):
2709
2710
class MyMixin:
2711
foo = Column(Integer, ForeignKey('bar.id'))
2712
2713
def go():
2714
class MyModel(Base, MyMixin):
2715
__tablename__ = 'foo'
2716
2717
assert_raises(sa.exc.InvalidRequestError, go)
2718
2719
class MyRelMixin:
2720
foo = relationship('Bar')
2721
2722
def go():
2723
class MyModel(Base, MyRelMixin):
2724
2725
__tablename__ = 'foo'
2726
2727
assert_raises(sa.exc.InvalidRequestError, go)
2728
2729
class MyDefMixin:
2730
foo = deferred(Column('foo', String))
2731
2732
def go():
2733
class MyModel(Base, MyDefMixin):
2734
__tablename__ = 'foo'
2735
2736
assert_raises(sa.exc.InvalidRequestError, go)
2737
2738
class MyCPropMixin:
2739
foo = column_property(Column('foo', String))
2740
2741
def go():
2742
class MyModel(Base, MyCPropMixin):
2743
__tablename__ = 'foo'
2744
2745
assert_raises(sa.exc.InvalidRequestError, go)
2746
2747
def test_table_name_inherited(self):
2748
2749
class MyMixin:
2750
@declared_attr
2751
def __tablename__(cls):
2752
return cls.__name__.lower()
2753
id = Column(Integer, primary_key=True)
2754
2755
class MyModel(Base, MyMixin):
2756
pass
2757
2758
eq_(MyModel.__table__.name, 'mymodel')
2759
2760
def test_classproperty_still_works(self):
2761
class MyMixin(object):
2762
@classproperty
2763
def __tablename__(cls):
2764
return cls.__name__.lower()
2765
id = Column(Integer, primary_key=True)
2766
2767
class MyModel(Base, MyMixin):
2768
__tablename__ = 'overridden'
2769
2770
eq_(MyModel.__table__.name, 'overridden')
2771
2772
def test_table_name_not_inherited(self):
2773
2774
class MyMixin:
2775
@declared_attr
2776
def __tablename__(cls):
2777
return cls.__name__.lower()
2778
id = Column(Integer, primary_key=True)
2779
2780
class MyModel(Base, MyMixin):
2781
__tablename__ = 'overridden'
2782
2783
eq_(MyModel.__table__.name, 'overridden')
2784
2785
def test_table_name_inheritance_order(self):
2786
2787
class MyMixin1:
2788
@declared_attr
2789
def __tablename__(cls):
2790
return cls.__name__.lower() + '1'
2791
2792
class MyMixin2:
2793
@declared_attr
2794
def __tablename__(cls):
2795
return cls.__name__.lower() + '2'
2796
2797
class MyModel(Base, MyMixin1, MyMixin2):
2798
id = Column(Integer, primary_key=True)
2799
2800
eq_(MyModel.__table__.name, 'mymodel1')
2801
2802
def test_table_name_dependent_on_subclass(self):
2803
2804
class MyHistoryMixin:
2805
@declared_attr
2806
def __tablename__(cls):
2807
return cls.parent_name + '_changelog'
2808
2809
class MyModel(Base, MyHistoryMixin):
2810
parent_name = 'foo'
2811
id = Column(Integer, primary_key=True)
2812
2813
eq_(MyModel.__table__.name, 'foo_changelog')
2814
2815
def test_table_args_inherited(self):
2816
2817
class MyMixin:
2818
__table_args__ = {'mysql_engine': 'InnoDB'}
2819
2820
class MyModel(Base, MyMixin):
2821
__tablename__ = 'test'
2822
id = Column(Integer, primary_key=True)
2823
2824
eq_(MyModel.__table__.kwargs, {'mysql_engine': 'InnoDB'})
2825
2826
def test_table_args_inherited_descriptor(self):
2827
2828
class MyMixin:
2829
@declared_attr
2830
def __table_args__(cls):
2831
return {'info': cls.__name__}
2832
2833
class MyModel(Base, MyMixin):
2834
__tablename__ = 'test'
2835
id = Column(Integer, primary_key=True)
2836
2837
eq_(MyModel.__table__.info, 'MyModel')
2838
2839
def test_table_args_inherited_single_table_inheritance(self):
2840
2841
class MyMixin:
2842
__table_args__ = {'mysql_engine': 'InnoDB'}
2843
2844
class General(Base, MyMixin):
2845
__tablename__ = 'test'
2846
id = Column(Integer, primary_key=True)
2847
type_ = Column(String(50))
2848
__mapper__args = {'polymorphic_on': type_}
2849
2850
class Specific(General):
2851
__mapper_args__ = {'polymorphic_identity': 'specific'}
2852
2853
assert Specific.__table__ is General.__table__
2854
eq_(General.__table__.kwargs, {'mysql_engine': 'InnoDB'})
2855
2856
def test_columns_single_table_inheritance(self):
2857
"""Test a column on a mixin with an alternate attribute name,
2858
mapped to a superclass and single-table inheritance subclass.
2859
The superclass table gets the column, the subclass shares
2860
the MapperProperty.
2861
2862
"""
2863
2864
class MyMixin(object):
2865
foo = Column('foo', Integer)
2866
bar = Column('bar_newname', Integer)
2867
2868
class General(Base, MyMixin):
2869
__tablename__ = 'test'
2870
id = Column(Integer, primary_key=True)
2871
type_ = Column(String(50))
2872
__mapper__args = {'polymorphic_on': type_}
2873
2874
class Specific(General):
2875
__mapper_args__ = {'polymorphic_identity': 'specific'}
2876
2877
assert General.bar.prop.columns[0] is General.__table__.c.bar_newname
2878
assert len(General.bar.prop.columns) == 1
2879
assert Specific.bar.prop is General.bar.prop
2880
2881
def test_columns_joined_table_inheritance(self):
2882
"""Test a column on a mixin with an alternate attribute name,
2883
mapped to a superclass and joined-table inheritance subclass.
2884
Both tables get the column, in the case of the subclass the two
2885
columns are joined under one MapperProperty.
2886
2887
"""
2888
2889
class MyMixin(object):
2890
foo = Column('foo', Integer)
2891
bar = Column('bar_newname', Integer)
2892
2893
class General(Base, MyMixin):
2894
__tablename__ = 'test'
2895
id = Column(Integer, primary_key=True)
2896
type_ = Column(String(50))
2897
__mapper__args = {'polymorphic_on': type_}
2898
2899
class Specific(General):
2900
__tablename__ = 'sub'
2901
id = Column(Integer, ForeignKey('test.id'), primary_key=True)
2902
__mapper_args__ = {'polymorphic_identity': 'specific'}
2903
2904
assert General.bar.prop.columns[0] is General.__table__.c.bar_newname
2905
assert len(General.bar.prop.columns) == 1
2906
assert Specific.bar.prop is not General.bar.prop
2907
assert len(Specific.bar.prop.columns) == 2
2908
assert Specific.bar.prop.columns[0] is Specific.__table__.c.bar_newname
2909
assert Specific.bar.prop.columns[1] is General.__table__.c.bar_newname
2910
2911
def test_column_join_checks_superclass_type(self):
2912
"""Test that the logic which joins subclass props to those
2913
of the superclass checks that the superclass property is a column.
2914
2915
"""
2916
2917
class General(Base):
2918
__tablename__ = 'test'
2919
id = Column(Integer, primary_key=True)
2920
general_id = Column(Integer, ForeignKey('test.id'))
2921
type_ = relationship("General")
2922
2923
class Specific(General):
2924
__tablename__ = 'sub'
2925
id = Column(Integer, ForeignKey('test.id'), primary_key=True)
2926
type_ = Column('foob', String(50))
2927
2928
assert isinstance(General.type_.property, sa.orm.RelationshipProperty)
2929
assert Specific.type_.property.columns[0] is Specific.__table__.c.foob
2930
2931
def test_column_join_checks_subclass_type(self):
2932
"""Test that the logic which joins subclass props to those
2933
of the superclass checks that the subclass property is a column.
2934
2935
"""
2936
2937
def go():
2938
class General(Base):
2939
__tablename__ = 'test'
2940
id = Column(Integer, primary_key=True)
2941
type_ = Column('foob', Integer)
2942
2943
class Specific(General):
2944
__tablename__ = 'sub'
2945
id = Column(Integer, ForeignKey('test.id'), primary_key=True)
2946
specific_id = Column(Integer, ForeignKey('sub.id'))
2947
type_ = relationship("Specific")
2948
assert_raises_message(
2949
sa.exc.ArgumentError, "column 'foob' conflicts with property", go
2950
)
2951
2952
def test_table_args_overridden(self):
2953
2954
class MyMixin:
2955
__table_args__ = {'mysql_engine': 'Foo'}
2956
2957
class MyModel(Base, MyMixin):
2958
__tablename__ = 'test'
2959
__table_args__ = {'mysql_engine': 'InnoDB'}
2960
id = Column(Integer, primary_key=True)
2961
2962
eq_(MyModel.__table__.kwargs, {'mysql_engine': 'InnoDB'})
2963
2964
def test_mapper_args_declared_attr(self):
2965
2966
class ComputedMapperArgs:
2967
@declared_attr
2968
def __mapper_args__(cls):
2969
if cls.__name__ == 'Person':
2970
return {'polymorphic_on': cls.discriminator}
2971
else:
2972
return {'polymorphic_identity': cls.__name__}
2973
2974
class Person(Base, ComputedMapperArgs):
2975
__tablename__ = 'people'
2976
id = Column(Integer, primary_key=True)
2977
discriminator = Column('type', String(50))
2978
2979
class Engineer(Person):
2980
pass
2981
2982
configure_mappers()
2983
assert class_mapper(Person).polymorphic_on \
2984
is Person.__table__.c.type
2985
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
2986
2987
def test_mapper_args_declared_attr_two(self):
2988
2989
# same as test_mapper_args_declared_attr, but we repeat
2990
# ComputedMapperArgs on both classes for no apparent reason.
2991
2992
class ComputedMapperArgs:
2993
@declared_attr
2994
def __mapper_args__(cls):
2995
if cls.__name__ == 'Person':
2996
return {'polymorphic_on': cls.discriminator}
2997
else:
2998
return {'polymorphic_identity': cls.__name__}
2999
3000
class Person(Base, ComputedMapperArgs):
3001
3002
__tablename__ = 'people'
3003
id = Column(Integer, primary_key=True)
3004
discriminator = Column('type', String(50))
3005
3006
class Engineer(Person, ComputedMapperArgs):
3007
pass
3008
3009
configure_mappers()
3010
assert class_mapper(Person).polymorphic_on \
3011
is Person.__table__.c.type
3012
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
3013
3014
def test_table_args_composite(self):
3015
3016
class MyMixin1:
3017
3018
__table_args__ = {'info': {'baz': 'bob'}}
3019
3020
class MyMixin2:
3021
3022
__table_args__ = {'info': {'foo': 'bar'}}
3023
3024
class MyModel(Base, MyMixin1, MyMixin2):
3025
3026
__tablename__ = 'test'
3027
3028
@declared_attr
3029
def __table_args__(self):
3030
info = {}
3031
args = dict(info=info)
3032
info.update(MyMixin1.__table_args__['info'])
3033
info.update(MyMixin2.__table_args__['info'])
3034
return args
3035
id = Column(Integer, primary_key=True)
3036
3037
eq_(MyModel.__table__.info, {'foo': 'bar', 'baz': 'bob'})
3038
3039
def test_mapper_args_inherited(self):
3040
3041
class MyMixin:
3042
3043
__mapper_args__ = {'always_refresh': True}
3044
3045
class MyModel(Base, MyMixin):
3046
3047
__tablename__ = 'test'
3048
id = Column(Integer, primary_key=True)
3049
3050
eq_(MyModel.__mapper__.always_refresh, True)
3051
3052
def test_mapper_args_inherited_descriptor(self):
3053
3054
class MyMixin:
3055
3056
@declared_attr
3057
def __mapper_args__(cls):
3058
3059
# tenuous, but illustrates the problem!
3060
3061
if cls.__name__ == 'MyModel':
3062
return dict(always_refresh=True)
3063
else:
3064
return dict(always_refresh=False)
3065
3066
class MyModel(Base, MyMixin):
3067
3068
__tablename__ = 'test'
3069
id = Column(Integer, primary_key=True)
3070
3071
eq_(MyModel.__mapper__.always_refresh, True)
3072
3073
def test_mapper_args_polymorphic_on_inherited(self):
3074
3075
class MyMixin:
3076
3077
type_ = Column(String(50))
3078
__mapper_args__ = {'polymorphic_on': type_}
3079
3080
class MyModel(Base, MyMixin):
3081
3082
__tablename__ = 'test'
3083
id = Column(Integer, primary_key=True)
3084
3085
col = MyModel.__mapper__.polymorphic_on
3086
eq_(col.name, 'type_')
3087
assert col.table is not None
3088
3089
def test_mapper_args_overridden(self):
3090
3091
class MyMixin:
3092
3093
__mapper_args__ = dict(always_refresh=True)
3094
3095
class MyModel(Base, MyMixin):
3096
3097
__tablename__ = 'test'
3098
__mapper_args__ = dict(always_refresh=False)
3099
id = Column(Integer, primary_key=True)
3100
3101
eq_(MyModel.__mapper__.always_refresh, False)
3102
3103
def test_mapper_args_composite(self):
3104
3105
class MyMixin1:
3106
3107
type_ = Column(String(50))
3108
__mapper_args__ = {'polymorphic_on': type_}
3109
3110
class MyMixin2:
3111
3112
__mapper_args__ = {'always_refresh': True}
3113
3114
class MyModel(Base, MyMixin1, MyMixin2):
3115
3116
__tablename__ = 'test'
3117
3118
@declared_attr
3119
def __mapper_args__(cls):
3120
args = {}
3121
args.update(MyMixin1.__mapper_args__)
3122
args.update(MyMixin2.__mapper_args__)
3123
if cls.__name__ != 'MyModel':
3124
args.pop('polymorphic_on')
3125
args['polymorphic_identity'] = cls.__name__
3126
3127
return args
3128
id = Column(Integer, primary_key=True)
3129
3130
class MySubModel(MyModel):
3131
pass
3132
3133
eq_(
3134
MyModel.__mapper__.polymorphic_on.name,
3135
'type_'
3136
)
3137
assert MyModel.__mapper__.polymorphic_on.table is not None
3138
eq_(MyModel.__mapper__.always_refresh, True)
3139
eq_(MySubModel.__mapper__.always_refresh, True)
3140
eq_(MySubModel.__mapper__.polymorphic_identity, 'MySubModel')
3141
3142
def test_mapper_args_property(self):
3143
class MyModel(Base):
3144
3145
@declared_attr
3146
def __tablename__(cls):
3147
return cls.__name__.lower()
3148
3149
@declared_attr
3150
def __table_args__(cls):
3151
return {'mysql_engine':'InnoDB'}
3152
3153
@declared_attr
3154
def __mapper_args__(cls):
3155
args = {}
3156
args['polymorphic_identity'] = cls.__name__
3157
return args
3158
id = Column(Integer, primary_key=True)
3159
3160
class MySubModel(MyModel):
3161
id = Column(Integer, ForeignKey('mymodel.id'), primary_key=True)
3162
3163
class MySubModel2(MyModel):
3164
__tablename__ = 'sometable'
3165
id = Column(Integer, ForeignKey('mymodel.id'), primary_key=True)
3166
3167
eq_(MyModel.__mapper__.polymorphic_identity, 'MyModel')
3168
eq_(MySubModel.__mapper__.polymorphic_identity, 'MySubModel')
3169
eq_(MyModel.__table__.kwargs['mysql_engine'], 'InnoDB')
3170
eq_(MySubModel.__table__.kwargs['mysql_engine'], 'InnoDB')
3171
eq_(MySubModel2.__table__.kwargs['mysql_engine'], 'InnoDB')
3172
eq_(MyModel.__table__.name, 'mymodel')
3173
eq_(MySubModel.__table__.name, 'mysubmodel')
3174
3175
def test_mapper_args_custom_base(self):
3176
"""test the @declared_attr approach from a custom base."""
3177
3178
class Base(object):
3179
@declared_attr
3180
def __tablename__(cls):
3181
return cls.__name__.lower()
3182
3183
@declared_attr
3184
def __table_args__(cls):
3185
return {'mysql_engine':'InnoDB'}
3186
3187
@declared_attr
3188
def id(self):
3189
return Column(Integer, primary_key=True)
3190
3191
Base = decl.declarative_base(cls=Base)
3192
3193
class MyClass(Base):
3194
pass
3195
3196
class MyOtherClass(Base):
3197
pass
3198
3199
eq_(MyClass.__table__.kwargs['mysql_engine'], 'InnoDB')
3200
eq_(MyClass.__table__.name, 'myclass')
3201
eq_(MyOtherClass.__table__.name, 'myotherclass')
3202
assert MyClass.__table__.c.id.table is MyClass.__table__
3203
assert MyOtherClass.__table__.c.id.table is MyOtherClass.__table__
3204
3205
def test_single_table_no_propagation(self):
3206
3207
class IdColumn:
3208
3209
id = Column(Integer, primary_key=True)
3210
3211
class Generic(Base, IdColumn):
3212
3213
__tablename__ = 'base'
3214
discriminator = Column('type', String(50))
3215
__mapper_args__ = dict(polymorphic_on=discriminator)
3216
value = Column(Integer())
3217
3218
class Specific(Generic):
3219
3220
__mapper_args__ = dict(polymorphic_identity='specific')
3221
3222
assert Specific.__table__ is Generic.__table__
3223
eq_(Generic.__table__.c.keys(), ['id', 'type', 'value'])
3224
assert class_mapper(Specific).polymorphic_on \
3225
is Generic.__table__.c.type
3226
eq_(class_mapper(Specific).polymorphic_identity, 'specific')
3227
3228
def test_joined_table_propagation(self):
3229
3230
class CommonMixin:
3231
3232
@declared_attr
3233
def __tablename__(cls):
3234
return cls.__name__.lower()
3235
__table_args__ = {'mysql_engine': 'InnoDB'}
3236
timestamp = Column(Integer)
3237
id = Column(Integer, primary_key=True)
3238
3239
class Generic(Base, CommonMixin):
3240
3241
discriminator = Column('python_type', String(50))
3242
__mapper_args__ = dict(polymorphic_on=discriminator)
3243
3244
class Specific(Generic):
3245
3246
__mapper_args__ = dict(polymorphic_identity='specific')
3247
id = Column(Integer, ForeignKey('generic.id'),
3248
primary_key=True)
3249
3250
eq_(Generic.__table__.name, 'generic')
3251
eq_(Specific.__table__.name, 'specific')
3252
eq_(Generic.__table__.c.keys(), ['timestamp', 'id',
3253
'python_type'])
3254
eq_(Specific.__table__.c.keys(), ['timestamp', 'id'])
3255
eq_(Generic.__table__.kwargs, {'mysql_engine': 'InnoDB'})
3256
eq_(Specific.__table__.kwargs, {'mysql_engine': 'InnoDB'})
3257
3258
def test_some_propagation(self):
3259
3260
class CommonMixin:
3261
3262
@declared_attr
3263
def __tablename__(cls):
3264
return cls.__name__.lower()
3265
__table_args__ = {'mysql_engine': 'InnoDB'}
3266
timestamp = Column(Integer)
3267
3268
class BaseType(Base, CommonMixin):
3269
3270
discriminator = Column('type', String(50))
3271
__mapper_args__ = dict(polymorphic_on=discriminator)
3272
id = Column(Integer, primary_key=True)
3273
value = Column(Integer())
3274
3275
class Single(BaseType):
3276
3277
__tablename__ = None
3278
__mapper_args__ = dict(polymorphic_identity='type1')
3279
3280
class Joined(BaseType):
3281
3282
__mapper_args__ = dict(polymorphic_identity='type2')
3283
id = Column(Integer, ForeignKey('basetype.id'),
3284
primary_key=True)
3285
3286
eq_(BaseType.__table__.name, 'basetype')
3287
eq_(BaseType.__table__.c.keys(), ['timestamp', 'type', 'id',
3288
'value'])
3289
eq_(BaseType.__table__.kwargs, {'mysql_engine': 'InnoDB'})
3290
assert Single.__table__ is BaseType.__table__
3291
eq_(Joined.__table__.name, 'joined')
3292
eq_(Joined.__table__.c.keys(), ['timestamp', 'id'])
3293
eq_(Joined.__table__.kwargs, {'mysql_engine': 'InnoDB'})
3294
3295
def test_non_propagating_mixin(self):
3296
3297
class NoJoinedTableNameMixin:
3298
3299
@declared_attr
3300
def __tablename__(cls):
3301
if decl.has_inherited_table(cls):
3302
return None
3303
return cls.__name__.lower()
3304
3305
class BaseType(Base, NoJoinedTableNameMixin):
3306
3307
discriminator = Column('type', String(50))
3308
__mapper_args__ = dict(polymorphic_on=discriminator)
3309
id = Column(Integer, primary_key=True)
3310
value = Column(Integer())
3311
3312
class Specific(BaseType):
3313
3314
__mapper_args__ = dict(polymorphic_identity='specific')
3315
3316
eq_(BaseType.__table__.name, 'basetype')
3317
eq_(BaseType.__table__.c.keys(), ['type', 'id', 'value'])
3318
assert Specific.__table__ is BaseType.__table__
3319
assert class_mapper(Specific).polymorphic_on \
3320
is BaseType.__table__.c.type
3321
eq_(class_mapper(Specific).polymorphic_identity, 'specific')
3322
3323
def test_non_propagating_mixin_used_for_joined(self):
3324
3325
class TableNameMixin:
3326
3327
@declared_attr
3328
def __tablename__(cls):
3329
if decl.has_inherited_table(cls) and TableNameMixin \
3330
not in cls.__bases__:
3331
return None
3332
return cls.__name__.lower()
3333
3334
class BaseType(Base, TableNameMixin):
3335
3336
discriminator = Column('type', String(50))
3337
__mapper_args__ = dict(polymorphic_on=discriminator)
3338
id = Column(Integer, primary_key=True)
3339
value = Column(Integer())
3340
3341
class Specific(BaseType, TableNameMixin):
3342
3343
__mapper_args__ = dict(polymorphic_identity='specific')
3344
id = Column(Integer, ForeignKey('basetype.id'),
3345
primary_key=True)
3346
3347
eq_(BaseType.__table__.name, 'basetype')
3348
eq_(BaseType.__table__.c.keys(), ['type', 'id', 'value'])
3349
eq_(Specific.__table__.name, 'specific')
3350
eq_(Specific.__table__.c.keys(), ['id'])
3351
3352
def test_single_back_propagate(self):
3353
3354
class ColumnMixin:
3355
3356
timestamp = Column(Integer)
3357
3358
class BaseType(Base):
3359
3360
__tablename__ = 'foo'
3361
discriminator = Column('type', String(50))
3362
__mapper_args__ = dict(polymorphic_on=discriminator)
3363
id = Column(Integer, primary_key=True)
3364
3365
class Specific(BaseType, ColumnMixin):
3366
3367
__mapper_args__ = dict(polymorphic_identity='specific')
3368
3369
eq_(BaseType.__table__.c.keys(), ['type', 'id', 'timestamp'])
3370
3371
def test_table_in_model_and_same_column_in_mixin(self):
3372
3373
class ColumnMixin:
3374
3375
data = Column(Integer)
3376
3377
class Model(Base, ColumnMixin):
3378
3379
__table__ = Table('foo', Base.metadata, Column('data',
3380
Integer), Column('id', Integer,
3381
primary_key=True))
3382
3383
model_col = Model.__table__.c.data
3384
mixin_col = ColumnMixin.data
3385
assert model_col is not mixin_col
3386
eq_(model_col.name, 'data')
3387
assert model_col.type.__class__ is mixin_col.type.__class__
3388
3389
def test_table_in_model_and_different_named_column_in_mixin(self):
3390
3391
class ColumnMixin:
3392
tada = Column(Integer)
3393
3394
def go():
3395
3396
class Model(Base, ColumnMixin):
3397
3398
__table__ = Table('foo', Base.metadata,
3399
Column('data',Integer),
3400
Column('id', Integer,primary_key=True))
3401
foo = relationship("Dest")
3402
3403
assert_raises_message(sa.exc.ArgumentError,
3404
"Can't add additional column 'tada' when "
3405
"specifying __table__", go)
3406
3407
def test_table_in_model_and_different_named_alt_key_column_in_mixin(self):
3408
3409
# here, the __table__ has a column 'tada'. We disallow
3410
# the add of the 'foobar' column, even though it's
3411
# keyed to 'tada'.
3412
3413
class ColumnMixin:
3414
tada = Column('foobar', Integer)
3415
3416
def go():
3417
3418
class Model(Base, ColumnMixin):
3419
3420
__table__ = Table('foo', Base.metadata,
3421
Column('data',Integer),
3422
Column('tada', Integer),
3423
Column('id', Integer,primary_key=True))
3424
foo = relationship("Dest")
3425
3426
assert_raises_message(sa.exc.ArgumentError,
3427
"Can't add additional column 'foobar' when "
3428
"specifying __table__", go)
3429
3430
def test_table_in_model_overrides_different_typed_column_in_mixin(self):
3431
3432
class ColumnMixin:
3433
3434
data = Column(String)
3435
3436
class Model(Base, ColumnMixin):
3437
3438
__table__ = Table('foo', Base.metadata, Column('data',
3439
Integer), Column('id', Integer,
3440
primary_key=True))
3441
3442
model_col = Model.__table__.c.data
3443
mixin_col = ColumnMixin.data
3444
assert model_col is not mixin_col
3445
eq_(model_col.name, 'data')
3446
assert model_col.type.__class__ is Integer
3447
3448
def test_mixin_column_ordering(self):
3449
3450
class Foo(object):
3451
3452
col1 = Column(Integer)
3453
col3 = Column(Integer)
3454
3455
class Bar(object):
3456
3457
col2 = Column(Integer)
3458
col4 = Column(Integer)
3459
3460
class Model(Base, Foo, Bar):
3461
3462
id = Column(Integer, primary_key=True)
3463
__tablename__ = 'model'
3464
3465
eq_(Model.__table__.c.keys(), ['col1', 'col3', 'col2', 'col4',
3466
'id'])
3467
3468
def test_honor_class_mro_one(self):
3469
class HasXMixin(object):
3470
@declared_attr
3471
def x(self):
3472
return Column(Integer)
3473
3474
class Parent(HasXMixin, Base):
3475
__tablename__ = 'parent'
3476
id = Column(Integer, primary_key=True)
3477
3478
class Child(Parent):
3479
__tablename__ = 'child'
3480
id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
3481
3482
assert "x" not in Child.__table__.c
3483
3484
def test_honor_class_mro_two(self):
3485
class HasXMixin(object):
3486
@declared_attr
3487
def x(self):
3488
return Column(Integer)
3489
3490
class Parent(HasXMixin, Base):
3491
__tablename__ = 'parent'
3492
id = Column(Integer, primary_key=True)
3493
def x(self):
3494
return "hi"
3495
3496
class C(Parent):
3497
__tablename__ = 'c'
3498
id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
3499
3500
assert C().x() == 'hi'
3501
3502
3503
class DeclarativeMixinPropertyTest(DeclarativeTestBase):
3504
3505
def test_column_property(self):
3506
3507
class MyMixin(object):
3508
3509
@declared_attr
3510
def prop_hoho(cls):
3511
return column_property(Column('prop', String(50)))
3512
3513
class MyModel(Base, MyMixin):
3514
3515
__tablename__ = 'test'
3516
id = Column(Integer, primary_key=True,
3517
test_needs_autoincrement=True)
3518
3519
class MyOtherModel(Base, MyMixin):
3520
3521
__tablename__ = 'othertest'
3522
id = Column(Integer, primary_key=True,
3523
test_needs_autoincrement=True)
3524
3525
assert MyModel.__table__.c.prop is not None
3526
assert MyOtherModel.__table__.c.prop is not None
3527
assert MyModel.__table__.c.prop \
3528
is not MyOtherModel.__table__.c.prop
3529
assert MyModel.prop_hoho.property.columns \
3530
== [MyModel.__table__.c.prop]
3531
assert MyOtherModel.prop_hoho.property.columns \
3532
== [MyOtherModel.__table__.c.prop]
3533
assert MyModel.prop_hoho.property \
3534
is not MyOtherModel.prop_hoho.property
3535
Base.metadata.create_all()
3536
sess = create_session()
3537
m1, m2 = MyModel(prop_hoho='foo'), MyOtherModel(prop_hoho='bar')
3538
sess.add_all([m1, m2])
3539
sess.flush()
3540
eq_(sess.query(MyModel).filter(MyModel.prop_hoho == 'foo'
3541
).one(), m1)
3542
eq_(sess.query(MyOtherModel).filter(MyOtherModel.prop_hoho
3543
== 'bar').one(), m2)
3544
3545
def test_doc(self):
3546
"""test documentation transfer.
3547
3548
the documentation situation with @declared_attr is problematic.
3549
at least see if mapped subclasses get the doc.
3550
3551
"""
3552
3553
class MyMixin(object):
3554
3555
@declared_attr
3556
def type_(cls):
3557
"""this is a document."""
3558
3559
return Column(String(50))
3560
3561
@declared_attr
3562
def t2(cls):
3563
"""this is another document."""
3564
3565
return column_property(Column(String(50)))
3566
3567
class MyModel(Base, MyMixin):
3568
3569
__tablename__ = 'test'
3570
id = Column(Integer, primary_key=True)
3571
3572
configure_mappers()
3573
eq_(MyModel.type_.__doc__, """this is a document.""")
3574
eq_(MyModel.t2.__doc__, """this is another document.""")
3575
3576
def test_column_in_mapper_args(self):
3577
3578
class MyMixin(object):
3579
3580
@declared_attr
3581
def type_(cls):
3582
return Column(String(50))
3583
__mapper_args__ = {'polymorphic_on': type_}
3584
3585
class MyModel(Base, MyMixin):
3586
3587
__tablename__ = 'test'
3588
id = Column(Integer, primary_key=True)
3589
3590
configure_mappers()
3591
col = MyModel.__mapper__.polymorphic_on
3592
eq_(col.name, 'type_')
3593
assert col.table is not None
3594
3595
def test_deferred(self):
3596
3597
class MyMixin(object):
3598
3599
@declared_attr
3600
def data(cls):
3601
return deferred(Column('data', String(50)))
3602
3603
class MyModel(Base, MyMixin):
3604
3605
__tablename__ = 'test'
3606
id = Column(Integer, primary_key=True,
3607
test_needs_autoincrement=True)
3608
3609
Base.metadata.create_all()
3610
sess = create_session()
3611
sess.add_all([MyModel(data='d1'), MyModel(data='d2')])
3612
sess.flush()
3613
sess.expunge_all()
3614
d1, d2 = sess.query(MyModel).order_by(MyModel.data)
3615
assert 'data' not in d1.__dict__
3616
assert d1.data == 'd1'
3617
assert 'data' in d1.__dict__
3618
3619
def _test_relationship(self, usestring):
3620
3621
class RefTargetMixin(object):
3622
3623
@declared_attr
3624
def target_id(cls):
3625
return Column('target_id', ForeignKey('target.id'))
3626
if usestring:
3627
3628
@declared_attr
3629
def target(cls):
3630
return relationship('Target',
3631
primaryjoin='Target.id==%s.target_id'
3632
% cls.__name__)
3633
else:
3634
3635
@declared_attr
3636
def target(cls):
3637
return relationship('Target')
3638
3639
class Foo(Base, RefTargetMixin):
3640
3641
__tablename__ = 'foo'
3642
id = Column(Integer, primary_key=True,
3643
test_needs_autoincrement=True)
3644
3645
class Bar(Base, RefTargetMixin):
3646
3647
__tablename__ = 'bar'
3648
id = Column(Integer, primary_key=True,
3649
test_needs_autoincrement=True)
3650
3651
class Target(Base):
3652
3653
__tablename__ = 'target'
3654
id = Column(Integer, primary_key=True,
3655
test_needs_autoincrement=True)
3656
3657
Base.metadata.create_all()
3658
sess = create_session()
3659
t1, t2 = Target(), Target()
3660
f1, f2, b1 = Foo(target=t1), Foo(target=t2), Bar(target=t1)
3661
sess.add_all([f1, f2, b1])
3662
sess.flush()
3663
eq_(sess.query(Foo).filter(Foo.target == t2).one(), f2)
3664
eq_(sess.query(Bar).filter(Bar.target == t2).first(), None)
3665
sess.expire_all()
3666
eq_(f1.target, t1)
3667
3668
def test_relationship(self):
3669
self._test_relationship(False)
3670
3671
def test_relationship_primryjoin(self):
3672
self._test_relationship(True)
3673
Loggerhead is a web-based interface for Breezy
Version: 2.0.1