← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/ext/declarative/test_basic.py

  • 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 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/ext/declarative/test_basic.py
 
1
 
 
2
from sqlalchemy.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 sqlalchemy import testing
 
8
from sqlalchemy import MetaData, Integer, String, ForeignKey, \
 
9
    ForeignKeyConstraint, Index
 
10
from sqlalchemy.testing.schema import Table, Column
 
11
from sqlalchemy.orm import relationship, create_session, class_mapper, \
 
12
    joinedload, configure_mappers, backref, clear_mappers, \
 
13
    deferred, column_property, composite,\
 
14
    Session
 
15
from sqlalchemy.testing import eq_
 
16
from sqlalchemy.util import classproperty
 
17
from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \
 
18
    ConcreteBase, synonym_for
 
19
from sqlalchemy.testing import fixtures
 
20
from sqlalchemy.testing.util import gc_collect
 
21
 
 
22
Base = None
 
23
 
 
24
class DeclarativeTestBase(fixtures.TestBase,
 
25
                            testing.AssertsExecutionResults,
 
26
                            testing.AssertsCompiledSQL):
 
27
    __dialect__ = 'default'
 
28
    def setup(self):
 
29
        global Base
 
30
        Base = decl.declarative_base(testing.db)
 
31
 
 
32
    def teardown(self):
 
33
        Session.close_all()
 
34
        clear_mappers()
 
35
        Base.metadata.drop_all()
 
36
 
 
37
class DeclarativeTest(DeclarativeTestBase):
 
38
    def test_basic(self):
 
39
        class User(Base, fixtures.ComparableEntity):
 
40
            __tablename__ = 'users'
 
41
 
 
42
            id = Column('id', Integer, primary_key=True,
 
43
                                        test_needs_autoincrement=True)
 
44
            name = Column('name', String(50))
 
45
            addresses = relationship("Address", backref="user")
 
46
 
 
47
        class Address(Base, fixtures.ComparableEntity):
 
48
            __tablename__ = 'addresses'
 
49
 
 
50
            id = Column(Integer, primary_key=True,
 
51
                                        test_needs_autoincrement=True)
 
52
            email = Column(String(50), key='_email')
 
53
            user_id = Column('user_id', Integer, ForeignKey('users.id'),
 
54
                             key='_user_id')
 
55
 
 
56
        Base.metadata.create_all()
 
57
 
 
58
        eq_(Address.__table__.c['id'].name, 'id')
 
59
        eq_(Address.__table__.c['_email'].name, 'email')
 
60
        eq_(Address.__table__.c['_user_id'].name, 'user_id')
 
61
 
 
62
        u1 = User(name='u1', addresses=[
 
63
            Address(email='one'),
 
64
            Address(email='two'),
 
65
        ])
 
66
        sess = create_session()
 
67
        sess.add(u1)
 
68
        sess.flush()
 
69
        sess.expunge_all()
 
70
 
 
71
        eq_(sess.query(User).all(), [User(name='u1', addresses=[
 
72
            Address(email='one'),
 
73
            Address(email='two'),
 
74
        ])])
 
75
 
 
76
        a1 = sess.query(Address).filter(Address.email == 'two').one()
 
77
        eq_(a1, Address(email='two'))
 
78
        eq_(a1.user, User(name='u1'))
 
79
 
 
80
    def test_no_table(self):
 
81
        def go():
 
82
            class User(Base):
 
83
                id = Column('id', Integer, primary_key=True)
 
84
 
 
85
        assert_raises_message(sa.exc.InvalidRequestError,
 
86
                              'does not have a __table__', go)
 
87
 
 
88
    def test_table_args_empty_dict(self):
 
89
 
 
90
        class MyModel(Base):
 
91
            __tablename__ = 'test'
 
92
            id = Column(Integer, primary_key=True)
 
93
            __table_args__ = {}
 
94
 
 
95
    def test_table_args_empty_tuple(self):
 
96
 
 
97
        class MyModel(Base):
 
98
            __tablename__ = 'test'
 
99
            id = Column(Integer, primary_key=True)
 
100
            __table_args__ = ()
 
101
 
 
102
    def test_cant_add_columns(self):
 
103
        t = Table('t', Base.metadata, Column('id', Integer,
 
104
                  primary_key=True), Column('data', String))
 
105
 
 
106
        def go():
 
107
            class User(Base):
 
108
                __table__ = t
 
109
                foo = Column(Integer, primary_key=True)
 
110
 
 
111
        # can't specify new columns not already in the table
 
112
 
 
113
        assert_raises_message(sa.exc.ArgumentError,
 
114
                              "Can't add additional column 'foo' when "
 
115
                              "specifying __table__", go)
 
116
 
 
117
        # regular re-mapping works tho
 
118
 
 
119
        class Bar(Base):
 
120
            __table__ = t
 
121
            some_data = t.c.data
 
122
 
 
123
        assert class_mapper(Bar).get_property('some_data').columns[0] \
 
124
            is t.c.data
 
125
 
 
126
    def test_difficult_class(self):
 
127
        """test no getattr() errors with a customized class"""
 
128
 
 
129
        # metaclass to mock the way zope.interface breaks getattr()
 
130
        class BrokenMeta(type):
 
131
            def __getattribute__(self, attr):
 
132
                if attr == 'xyzzy':
 
133
                    raise AttributeError, 'xyzzy'
 
134
                else:
 
135
                    return object.__getattribute__(self,attr)
 
136
 
 
137
        # even though this class has an xyzzy attribute, getattr(cls,"xyzzy")
 
138
        # fails
 
139
        class BrokenParent(object):
 
140
            __metaclass__ = BrokenMeta
 
141
            xyzzy = "magic"
 
142
 
 
143
        # _as_declarative() inspects obj.__class__.__bases__
 
144
        class User(BrokenParent,fixtures.ComparableEntity):
 
145
            __tablename__ = 'users'
 
146
            id = Column('id', Integer, primary_key=True,
 
147
                test_needs_autoincrement=True)
 
148
            name = Column('name', String(50))
 
149
 
 
150
        decl.instrument_declarative(User,{},Base.metadata)
 
151
 
 
152
    def test_reserved_identifiers(self):
 
153
        def go1():
 
154
            class User1(Base):
 
155
                __tablename__ = 'user1'
 
156
                id = Column(Integer, primary_key=True)
 
157
                metadata = Column(Integer)
 
158
 
 
159
        def go2():
 
160
            class User2(Base):
 
161
                __tablename__ = 'user2'
 
162
                id = Column(Integer, primary_key=True)
 
163
                metadata = relationship("Address")
 
164
 
 
165
        for go in (go1, go2):
 
166
            assert_raises_message(
 
167
                exc.InvalidRequestError,
 
168
                "Attribute name 'metadata' is reserved "
 
169
                "for the MetaData instance when using a "
 
170
                "declarative base class.",
 
171
                go
 
172
            )
 
173
 
 
174
    def test_undefer_column_name(self):
 
175
        # TODO: not sure if there was an explicit
 
176
        # test for this elsewhere
 
177
        foo = Column(Integer)
 
178
        eq_(str(foo), '(no name)')
 
179
        eq_(foo.key, None)
 
180
        eq_(foo.name, None)
 
181
        decl.base._undefer_column_name('foo', foo)
 
182
        eq_(str(foo), 'foo')
 
183
        eq_(foo.key, 'foo')
 
184
        eq_(foo.name, 'foo')
 
185
 
 
186
    def test_recompile_on_othermapper(self):
 
187
        """declarative version of the same test in mappers.py"""
 
188
 
 
189
        from sqlalchemy.orm import mapperlib
 
190
 
 
191
        class User(Base):
 
192
            __tablename__ = 'users'
 
193
 
 
194
            id = Column('id', Integer, primary_key=True)
 
195
            name = Column('name', String(50))
 
196
 
 
197
        class Address(Base):
 
198
            __tablename__ = 'addresses'
 
199
 
 
200
            id = Column('id', Integer, primary_key=True)
 
201
            email = Column('email', String(50))
 
202
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
203
            user = relationship("User", primaryjoin=user_id == User.id,
 
204
                            backref="addresses")
 
205
 
 
206
        assert mapperlib._new_mappers is True
 
207
        u = User()
 
208
        assert User.addresses
 
209
        assert mapperlib._new_mappers is False
 
210
 
 
211
    def test_string_dependency_resolution(self):
 
212
        from sqlalchemy.sql import desc
 
213
 
 
214
        class User(Base, fixtures.ComparableEntity):
 
215
 
 
216
            __tablename__ = 'users'
 
217
            id = Column(Integer, primary_key=True,
 
218
                        test_needs_autoincrement=True)
 
219
            name = Column(String(50))
 
220
            addresses = relationship('Address',
 
221
                    order_by='desc(Address.email)',
 
222
                    primaryjoin='User.id==Address.user_id',
 
223
                    foreign_keys='[Address.user_id]',
 
224
                    backref=backref('user',
 
225
                    primaryjoin='User.id==Address.user_id',
 
226
                    foreign_keys='[Address.user_id]'))
 
227
 
 
228
        class Address(Base, fixtures.ComparableEntity):
 
229
 
 
230
            __tablename__ = 'addresses'
 
231
            id = Column(Integer, primary_key=True,
 
232
                        test_needs_autoincrement=True)
 
233
            email = Column(String(50))
 
234
            user_id = Column(Integer)  # note no foreign key
 
235
 
 
236
        Base.metadata.create_all()
 
237
        sess = create_session()
 
238
        u1 = User(name='ed', addresses=[Address(email='abc'),
 
239
                  Address(email='def'), Address(email='xyz')])
 
240
        sess.add(u1)
 
241
        sess.flush()
 
242
        sess.expunge_all()
 
243
        eq_(sess.query(User).filter(User.name == 'ed').one(),
 
244
            User(name='ed', addresses=[Address(email='xyz'),
 
245
            Address(email='def'), Address(email='abc')]))
 
246
 
 
247
        class Foo(Base, fixtures.ComparableEntity):
 
248
 
 
249
            __tablename__ = 'foo'
 
250
            id = Column(Integer, primary_key=True)
 
251
            rel = relationship('User',
 
252
                               primaryjoin='User.addresses==Foo.id')
 
253
 
 
254
        assert_raises_message(exc.InvalidRequestError,
 
255
                              "'addresses' is not an instance of "
 
256
                              "ColumnProperty", configure_mappers)
 
257
 
 
258
    def test_string_dependency_resolution_synonym(self):
 
259
        from sqlalchemy.sql import desc
 
260
 
 
261
        class User(Base, fixtures.ComparableEntity):
 
262
 
 
263
            __tablename__ = 'users'
 
264
            id = Column(Integer, primary_key=True,
 
265
                        test_needs_autoincrement=True)
 
266
            name = Column(String(50))
 
267
 
 
268
        Base.metadata.create_all()
 
269
        sess = create_session()
 
270
        u1 = User(name='ed')
 
271
        sess.add(u1)
 
272
        sess.flush()
 
273
        sess.expunge_all()
 
274
        eq_(sess.query(User).filter(User.name == 'ed').one(),
 
275
            User(name='ed'))
 
276
 
 
277
        class Foo(Base, fixtures.ComparableEntity):
 
278
 
 
279
            __tablename__ = 'foo'
 
280
            id = Column(Integer, primary_key=True)
 
281
            _user_id = Column(Integer) 
 
282
            rel = relationship('User',
 
283
                               uselist=False,
 
284
                               foreign_keys=[User.id],
 
285
                               primaryjoin='Foo.user_id==User.id')
 
286
 
 
287
            @synonym_for('_user_id')
 
288
            @property
 
289
            def user_id(self):
 
290
                return self._user_id
 
291
 
 
292
        foo = Foo()
 
293
        foo.rel = u1
 
294
        assert foo.rel == u1
 
295
 
 
296
    def test_string_dependency_resolution_orm_descriptor(self):
 
297
        from sqlalchemy.ext.hybrid import hybrid_property
 
298
 
 
299
        class User(Base):
 
300
            __tablename__ = 'user'
 
301
            id = Column(Integer, primary_key=True)
 
302
            firstname = Column(String(50))
 
303
            lastname = Column(String(50))
 
304
            game_id = Column(Integer, ForeignKey('game.id'))
 
305
 
 
306
            @hybrid_property
 
307
            def fullname(self):
 
308
                return self.firstname + " " + self.lastname
 
309
 
 
310
        class Game(Base):
 
311
            __tablename__ = 'game'
 
312
            id = Column(Integer, primary_key=True)
 
313
            name = Column(String(50))
 
314
            users = relationship("User", order_by="User.fullname")
 
315
 
 
316
        s = Session()
 
317
        self.assert_compile(
 
318
            s.query(Game).options(joinedload(Game.users)),
 
319
            "SELECT game.id AS game_id, game.name AS game_name, "
 
320
            "user_1.id AS user_1_id, user_1.firstname AS user_1_firstname, "
 
321
            "user_1.lastname AS user_1_lastname, "
 
322
            "user_1.game_id AS user_1_game_id "
 
323
            "FROM game LEFT OUTER JOIN \"user\" AS user_1 ON game.id = "
 
324
            "user_1.game_id ORDER BY "
 
325
            "user_1.firstname || :firstname_1 || user_1.lastname"
 
326
        )
 
327
 
 
328
    def test_string_dependency_resolution_no_table(self):
 
329
 
 
330
        class User(Base, fixtures.ComparableEntity):
 
331
            __tablename__ = 'users'
 
332
            id = Column(Integer, primary_key=True,
 
333
                        test_needs_autoincrement=True)
 
334
            name = Column(String(50))
 
335
 
 
336
        class Bar(Base, fixtures.ComparableEntity):
 
337
            __tablename__ = 'bar'
 
338
            id = Column(Integer, primary_key=True)
 
339
            rel = relationship('User',
 
340
                               primaryjoin='User.id==Bar.__table__.id')
 
341
 
 
342
        assert_raises_message(exc.InvalidRequestError,
 
343
                              "does not have a mapped column named "
 
344
                              "'__table__'", configure_mappers)
 
345
 
 
346
    def test_string_w_pj_annotations(self):
 
347
 
 
348
        class User(Base, fixtures.ComparableEntity):
 
349
            __tablename__ = 'users'
 
350
            id = Column(Integer, primary_key=True,
 
351
                        test_needs_autoincrement=True)
 
352
            name = Column(String(50))
 
353
        class Address(Base, fixtures.ComparableEntity):
 
354
 
 
355
            __tablename__ = 'addresses'
 
356
            id = Column(Integer, primary_key=True,
 
357
                        test_needs_autoincrement=True)
 
358
            email = Column(String(50))
 
359
            user_id = Column(Integer)
 
360
            user = relationship("User",
 
361
                primaryjoin="remote(User.id)==foreign(Address.user_id)"
 
362
            )
 
363
 
 
364
        eq_(
 
365
            Address.user.property._join_condition.local_remote_pairs,
 
366
            [(Address.__table__.c.user_id, User.__table__.c.id)]
 
367
        )
 
368
 
 
369
    def test_string_dependency_resolution_no_magic(self):
 
370
        """test that full tinkery expressions work as written"""
 
371
 
 
372
        class User(Base, fixtures.ComparableEntity):
 
373
 
 
374
            __tablename__ = 'users'
 
375
            id = Column(Integer, primary_key=True)
 
376
            addresses = relationship('Address',
 
377
                    primaryjoin='User.id==Address.user_id.prop.columns['
 
378
                    '0]')
 
379
 
 
380
        class Address(Base, fixtures.ComparableEntity):
 
381
 
 
382
            __tablename__ = 'addresses'
 
383
            id = Column(Integer, primary_key=True)
 
384
            user_id = Column(Integer, ForeignKey('users.id'))
 
385
 
 
386
        configure_mappers()
 
387
        eq_(str(User.addresses.prop.primaryjoin),
 
388
            'users.id = addresses.user_id')
 
389
 
 
390
    def test_string_dependency_resolution_module_qualified(self):
 
391
        class User(Base, fixtures.ComparableEntity):
 
392
 
 
393
            __tablename__ = 'users'
 
394
            id = Column(Integer, primary_key=True)
 
395
            addresses = relationship('%s.Address' % __name__,
 
396
                    primaryjoin='%s.User.id==%s.Address.user_id.prop.columns['
 
397
                    '0]' % (__name__, __name__))
 
398
 
 
399
        class Address(Base, fixtures.ComparableEntity):
 
400
 
 
401
            __tablename__ = 'addresses'
 
402
            id = Column(Integer, primary_key=True)
 
403
            user_id = Column(Integer, ForeignKey('users.id'))
 
404
 
 
405
        configure_mappers()
 
406
        eq_(str(User.addresses.prop.primaryjoin),
 
407
            'users.id = addresses.user_id')
 
408
 
 
409
    def test_string_dependency_resolution_in_backref(self):
 
410
 
 
411
        class User(Base, fixtures.ComparableEntity):
 
412
 
 
413
            __tablename__ = 'users'
 
414
            id = Column(Integer, primary_key=True)
 
415
            name = Column(String(50))
 
416
            addresses = relationship('Address',
 
417
                    primaryjoin='User.id==Address.user_id',
 
418
                    backref='user')
 
419
 
 
420
        class Address(Base, fixtures.ComparableEntity):
 
421
 
 
422
            __tablename__ = 'addresses'
 
423
            id = Column(Integer, primary_key=True)
 
424
            email = Column(String(50))
 
425
            user_id = Column(Integer, ForeignKey('users.id'))
 
426
 
 
427
        configure_mappers()
 
428
        eq_(str(User.addresses.property.primaryjoin),
 
429
            str(Address.user.property.primaryjoin))
 
430
 
 
431
    def test_string_dependency_resolution_tables(self):
 
432
 
 
433
        class User(Base, fixtures.ComparableEntity):
 
434
 
 
435
            __tablename__ = 'users'
 
436
            id = Column(Integer, primary_key=True)
 
437
            name = Column(String(50))
 
438
            props = relationship('Prop', secondary='user_to_prop',
 
439
                                 primaryjoin='User.id==user_to_prop.c.u'
 
440
                                 'ser_id',
 
441
                                 secondaryjoin='user_to_prop.c.prop_id='
 
442
                                 '=Prop.id', backref='users')
 
443
 
 
444
        class Prop(Base, fixtures.ComparableEntity):
 
445
 
 
446
            __tablename__ = 'props'
 
447
            id = Column(Integer, primary_key=True)
 
448
            name = Column(String(50))
 
449
 
 
450
        user_to_prop = Table('user_to_prop', Base.metadata,
 
451
                             Column('user_id', Integer,
 
452
                             ForeignKey('users.id')), Column('prop_id',
 
453
                             Integer, ForeignKey('props.id')))
 
454
        configure_mappers()
 
455
        assert class_mapper(User).get_property('props').secondary \
 
456
            is user_to_prop
 
457
 
 
458
    def test_string_dependency_resolution_schemas(self):
 
459
        Base = decl.declarative_base()
 
460
 
 
461
        class User(Base):
 
462
 
 
463
            __tablename__ = 'users'
 
464
            __table_args__ = {'schema':'fooschema'}
 
465
 
 
466
            id = Column(Integer, primary_key=True)
 
467
            name = Column(String(50))
 
468
            props = relationship('Prop', secondary='fooschema.user_to_prop',
 
469
                         primaryjoin='User.id==fooschema.user_to_prop.c.user_id',
 
470
                         secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id',
 
471
                         backref='users')
 
472
 
 
473
        class Prop(Base):
 
474
 
 
475
            __tablename__ = 'props'
 
476
            __table_args__ = {'schema':'fooschema'}
 
477
 
 
478
            id = Column(Integer, primary_key=True)
 
479
            name = Column(String(50))
 
480
 
 
481
        user_to_prop = Table('user_to_prop', Base.metadata,
 
482
                     Column('user_id', Integer, ForeignKey('fooschema.users.id')),
 
483
                     Column('prop_id',Integer, ForeignKey('fooschema.props.id')),
 
484
                     schema='fooschema')
 
485
        configure_mappers()
 
486
 
 
487
        assert class_mapper(User).get_property('props').secondary \
 
488
            is user_to_prop
 
489
 
 
490
    def test_string_dependency_resolution_annotations(self):
 
491
        Base = decl.declarative_base()
 
492
 
 
493
        class Parent(Base):
 
494
            __tablename__ = 'parent'
 
495
            id = Column(Integer, primary_key=True)
 
496
            name = Column(String)
 
497
            children = relationship("Child",
 
498
                    primaryjoin="Parent.name==remote(foreign(func.lower(Child.name_upper)))"
 
499
                )
 
500
 
 
501
        class Child(Base):
 
502
            __tablename__ = 'child'
 
503
            id = Column(Integer, primary_key=True)
 
504
            name_upper = Column(String)
 
505
 
 
506
        configure_mappers()
 
507
        eq_(
 
508
            Parent.children.property._calculated_foreign_keys,
 
509
            set([Child.name_upper.property.columns[0]])
 
510
        )
 
511
 
 
512
    def test_shared_class_registry(self):
 
513
        reg = {}
 
514
        Base1 = decl.declarative_base(testing.db, class_registry=reg)
 
515
        Base2 = decl.declarative_base(testing.db, class_registry=reg)
 
516
 
 
517
        class A(Base1):
 
518
            __tablename__ = 'a'
 
519
            id = Column(Integer, primary_key=True)
 
520
 
 
521
        class B(Base2):
 
522
            __tablename__ = 'b'
 
523
            id = Column(Integer, primary_key=True)
 
524
            aid = Column(Integer, ForeignKey(A.id))
 
525
            as_ = relationship("A")
 
526
 
 
527
        assert B.as_.property.mapper.class_ is A
 
528
 
 
529
    def test_uncompiled_attributes_in_relationship(self):
 
530
 
 
531
        class Address(Base, fixtures.ComparableEntity):
 
532
 
 
533
            __tablename__ = 'addresses'
 
534
            id = Column(Integer, primary_key=True,
 
535
                        test_needs_autoincrement=True)
 
536
            email = Column(String(50))
 
537
            user_id = Column(Integer, ForeignKey('users.id'))
 
538
 
 
539
        class User(Base, fixtures.ComparableEntity):
 
540
 
 
541
            __tablename__ = 'users'
 
542
            id = Column(Integer, primary_key=True,
 
543
                        test_needs_autoincrement=True)
 
544
            name = Column(String(50))
 
545
            addresses = relationship('Address', order_by=Address.email,
 
546
                    foreign_keys=Address.user_id,
 
547
                    remote_side=Address.user_id)
 
548
 
 
549
        # get the mapper for User.   User mapper will compile,
 
550
        # "addresses" relationship will call upon Address.user_id for
 
551
        # its clause element.  Address.user_id is a _CompileOnAttr,
 
552
        # which then calls class_mapper(Address).  But !  We're already
 
553
        # "in compilation", but class_mapper(Address) needs to
 
554
        # initialize regardless, or COA's assertion fails and things
 
555
        # generally go downhill from there.
 
556
 
 
557
        class_mapper(User)
 
558
        Base.metadata.create_all()
 
559
        sess = create_session()
 
560
        u1 = User(name='ed', addresses=[Address(email='abc'),
 
561
                  Address(email='xyz'), Address(email='def')])
 
562
        sess.add(u1)
 
563
        sess.flush()
 
564
        sess.expunge_all()
 
565
        eq_(sess.query(User).filter(User.name == 'ed').one(),
 
566
            User(name='ed', addresses=[Address(email='abc'),
 
567
            Address(email='def'), Address(email='xyz')]))
 
568
 
 
569
    def test_nice_dependency_error(self):
 
570
 
 
571
        class User(Base):
 
572
 
 
573
            __tablename__ = 'users'
 
574
            id = Column('id', Integer, primary_key=True)
 
575
            addresses = relationship('Address')
 
576
 
 
577
        class Address(Base):
 
578
 
 
579
            __tablename__ = 'addresses'
 
580
            id = Column(Integer, primary_key=True)
 
581
            foo = sa.orm.column_property(User.id == 5)
 
582
 
 
583
        # this used to raise an error when accessing User.id but that's
 
584
        # no longer the case since we got rid of _CompileOnAttr.
 
585
 
 
586
        assert_raises(sa.exc.ArgumentError, configure_mappers)
 
587
 
 
588
    def test_nice_dependency_error_works_with_hasattr(self):
 
589
 
 
590
        class User(Base):
 
591
 
 
592
            __tablename__ = 'users'
 
593
            id = Column('id', Integer, primary_key=True)
 
594
            addresses = relationship('Address')
 
595
 
 
596
        # hasattr() on a compile-loaded attribute
 
597
        try:
 
598
            hasattr(User.addresses, 'property')
 
599
        except exc.InvalidRequestError:
 
600
            assert sa.util.compat.py32
 
601
 
 
602
        # the exception is preserved.  Remains the
 
603
        # same through repeated calls.
 
604
        for i in range(3):
 
605
            assert_raises_message(sa.exc.InvalidRequestError,
 
606
                            "^One or more mappers failed to initialize - "
 
607
                            "can't proceed with initialization of other "
 
608
                            "mappers.  Original exception was: When initializing.*",
 
609
                            configure_mappers)
 
610
 
 
611
    def test_custom_base(self):
 
612
        class MyBase(object):
 
613
            def foobar(self):
 
614
                return "foobar"
 
615
        Base = decl.declarative_base(cls=MyBase)
 
616
        assert hasattr(Base, 'metadata')
 
617
        assert Base().foobar() == "foobar"
 
618
 
 
619
    def test_uses_get_on_class_col_fk(self):
 
620
 
 
621
        # test [ticket:1492]
 
622
 
 
623
        class Master(Base):
 
624
 
 
625
            __tablename__ = 'master'
 
626
            id = Column(Integer, primary_key=True,
 
627
                        test_needs_autoincrement=True)
 
628
 
 
629
        class Detail(Base):
 
630
 
 
631
            __tablename__ = 'detail'
 
632
            id = Column(Integer, primary_key=True,
 
633
                        test_needs_autoincrement=True)
 
634
            master_id = Column(None, ForeignKey(Master.id))
 
635
            master = relationship(Master)
 
636
 
 
637
        Base.metadata.create_all()
 
638
        configure_mappers()
 
639
        assert class_mapper(Detail).get_property('master'
 
640
                ).strategy.use_get
 
641
        m1 = Master()
 
642
        d1 = Detail(master=m1)
 
643
        sess = create_session()
 
644
        sess.add(d1)
 
645
        sess.flush()
 
646
        sess.expunge_all()
 
647
        d1 = sess.query(Detail).first()
 
648
        m1 = sess.query(Master).first()
 
649
 
 
650
        def go():
 
651
            assert d1.master
 
652
 
 
653
        self.assert_sql_count(testing.db, go, 0)
 
654
 
 
655
    def test_index_doesnt_compile(self):
 
656
        class User(Base):
 
657
            __tablename__ = 'users'
 
658
            id = Column('id', Integer, primary_key=True)
 
659
            name = Column('name', String(50))
 
660
            error = relationship("Address")
 
661
 
 
662
        i = Index('my_index', User.name)
 
663
 
 
664
        # compile fails due to the nonexistent Addresses relationship
 
665
        assert_raises(sa.exc.InvalidRequestError, configure_mappers)
 
666
 
 
667
        # index configured
 
668
        assert i in User.__table__.indexes
 
669
        assert User.__table__.c.id not in set(i.columns)
 
670
        assert User.__table__.c.name in set(i.columns)
 
671
 
 
672
        # tables create fine
 
673
        Base.metadata.create_all()
 
674
 
 
675
    def test_add_prop(self):
 
676
 
 
677
        class User(Base, fixtures.ComparableEntity):
 
678
 
 
679
            __tablename__ = 'users'
 
680
            id = Column('id', Integer, primary_key=True,
 
681
                        test_needs_autoincrement=True)
 
682
 
 
683
        User.name = Column('name', String(50))
 
684
        User.addresses = relationship('Address', backref='user')
 
685
 
 
686
        class Address(Base, fixtures.ComparableEntity):
 
687
 
 
688
            __tablename__ = 'addresses'
 
689
            id = Column(Integer, primary_key=True,
 
690
                        test_needs_autoincrement=True)
 
691
 
 
692
        Address.email = Column(String(50), key='_email')
 
693
        Address.user_id = Column('user_id', Integer,
 
694
                                 ForeignKey('users.id'), key='_user_id')
 
695
        Base.metadata.create_all()
 
696
        eq_(Address.__table__.c['id'].name, 'id')
 
697
        eq_(Address.__table__.c['_email'].name, 'email')
 
698
        eq_(Address.__table__.c['_user_id'].name, 'user_id')
 
699
        u1 = User(name='u1', addresses=[Address(email='one'),
 
700
                  Address(email='two')])
 
701
        sess = create_session()
 
702
        sess.add(u1)
 
703
        sess.flush()
 
704
        sess.expunge_all()
 
705
        eq_(sess.query(User).all(), [User(name='u1',
 
706
            addresses=[Address(email='one'), Address(email='two')])])
 
707
        a1 = sess.query(Address).filter(Address.email == 'two').one()
 
708
        eq_(a1, Address(email='two'))
 
709
        eq_(a1.user, User(name='u1'))
 
710
 
 
711
    def test_eager_order_by(self):
 
712
 
 
713
        class Address(Base, fixtures.ComparableEntity):
 
714
 
 
715
            __tablename__ = 'addresses'
 
716
            id = Column('id', Integer, primary_key=True,
 
717
                        test_needs_autoincrement=True)
 
718
            email = Column('email', String(50))
 
719
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
720
 
 
721
        class User(Base, fixtures.ComparableEntity):
 
722
 
 
723
            __tablename__ = 'users'
 
724
            id = Column('id', Integer, primary_key=True,
 
725
                        test_needs_autoincrement=True)
 
726
            name = Column('name', String(50))
 
727
            addresses = relationship('Address', order_by=Address.email)
 
728
 
 
729
        Base.metadata.create_all()
 
730
        u1 = User(name='u1', addresses=[Address(email='two'),
 
731
                  Address(email='one')])
 
732
        sess = create_session()
 
733
        sess.add(u1)
 
734
        sess.flush()
 
735
        sess.expunge_all()
 
736
        eq_(sess.query(User).options(joinedload(User.addresses)).all(),
 
737
            [User(name='u1', addresses=[Address(email='one'),
 
738
            Address(email='two')])])
 
739
 
 
740
    def test_order_by_multi(self):
 
741
 
 
742
        class Address(Base, fixtures.ComparableEntity):
 
743
 
 
744
            __tablename__ = 'addresses'
 
745
            id = Column('id', Integer, primary_key=True,
 
746
                        test_needs_autoincrement=True)
 
747
            email = Column('email', String(50))
 
748
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
749
 
 
750
        class User(Base, fixtures.ComparableEntity):
 
751
 
 
752
            __tablename__ = 'users'
 
753
            id = Column('id', Integer, primary_key=True,
 
754
                        test_needs_autoincrement=True)
 
755
            name = Column('name', String(50))
 
756
            addresses = relationship('Address',
 
757
                    order_by=(Address.email, Address.id))
 
758
 
 
759
        Base.metadata.create_all()
 
760
        u1 = User(name='u1', addresses=[Address(email='two'),
 
761
                  Address(email='one')])
 
762
        sess = create_session()
 
763
        sess.add(u1)
 
764
        sess.flush()
 
765
        sess.expunge_all()
 
766
        u = sess.query(User).filter(User.name == 'u1').one()
 
767
        a = u.addresses
 
768
 
 
769
    def test_as_declarative(self):
 
770
 
 
771
        class User(fixtures.ComparableEntity):
 
772
 
 
773
            __tablename__ = 'users'
 
774
            id = Column('id', Integer, primary_key=True,
 
775
                        test_needs_autoincrement=True)
 
776
            name = Column('name', String(50))
 
777
            addresses = relationship('Address', backref='user')
 
778
 
 
779
        class Address(fixtures.ComparableEntity):
 
780
 
 
781
            __tablename__ = 'addresses'
 
782
            id = Column('id', Integer, primary_key=True,
 
783
                        test_needs_autoincrement=True)
 
784
            email = Column('email', String(50))
 
785
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
786
 
 
787
        reg = {}
 
788
        decl.instrument_declarative(User, reg, Base.metadata)
 
789
        decl.instrument_declarative(Address, reg, Base.metadata)
 
790
        Base.metadata.create_all()
 
791
        u1 = User(name='u1', addresses=[Address(email='one'),
 
792
                  Address(email='two')])
 
793
        sess = create_session()
 
794
        sess.add(u1)
 
795
        sess.flush()
 
796
        sess.expunge_all()
 
797
        eq_(sess.query(User).all(), [User(name='u1',
 
798
            addresses=[Address(email='one'), Address(email='two')])])
 
799
 
 
800
    def test_custom_mapper_attribute(self):
 
801
 
 
802
        def mymapper(cls, tbl, **kwargs):
 
803
            m = sa.orm.mapper(cls, tbl, **kwargs)
 
804
            m.CHECK = True
 
805
            return m
 
806
 
 
807
        base = decl.declarative_base()
 
808
 
 
809
        class Foo(base):
 
810
            __tablename__ = 'foo'
 
811
            __mapper_cls__ = mymapper
 
812
            id = Column(Integer, primary_key=True)
 
813
 
 
814
        eq_(Foo.__mapper__.CHECK, True)
 
815
 
 
816
    def test_custom_mapper_argument(self):
 
817
 
 
818
        def mymapper(cls, tbl, **kwargs):
 
819
            m = sa.orm.mapper(cls, tbl, **kwargs)
 
820
            m.CHECK = True
 
821
            return m
 
822
 
 
823
        base = decl.declarative_base(mapper=mymapper)
 
824
 
 
825
        class Foo(base):
 
826
            __tablename__ = 'foo'
 
827
            id = Column(Integer, primary_key=True)
 
828
 
 
829
        eq_(Foo.__mapper__.CHECK, True)
 
830
 
 
831
    @testing.emits_warning('Ignoring declarative-like tuple value of '
 
832
                           'attribute id')
 
833
    def test_oops(self):
 
834
 
 
835
        def define():
 
836
 
 
837
            class User(Base, fixtures.ComparableEntity):
 
838
 
 
839
                __tablename__ = 'users'
 
840
                id = Column('id', Integer, primary_key=True),
 
841
                name = Column('name', String(50))
 
842
 
 
843
            assert False
 
844
 
 
845
        assert_raises_message(sa.exc.ArgumentError,
 
846
                              'Mapper Mapper|User|users could not '
 
847
                              'assemble any primary key', define)
 
848
 
 
849
    def test_table_args_no_dict(self):
 
850
 
 
851
        class Foo1(Base):
 
852
 
 
853
            __tablename__ = 'foo'
 
854
            __table_args__ = ForeignKeyConstraint(['id'], ['foo.bar']),
 
855
            id = Column('id', Integer, primary_key=True)
 
856
            bar = Column('bar', Integer)
 
857
 
 
858
        assert Foo1.__table__.c.id.references(Foo1.__table__.c.bar)
 
859
 
 
860
    def test_table_args_type(self):
 
861
        def err():
 
862
            class Foo1(Base):
 
863
 
 
864
                __tablename__ = 'foo'
 
865
                __table_args__ = ForeignKeyConstraint(['id'], ['foo.id'
 
866
                        ])
 
867
                id = Column('id', Integer, primary_key=True)
 
868
        assert_raises_message(sa.exc.ArgumentError,
 
869
                              '__table_args__ value must be a tuple, ', err)
 
870
 
 
871
    def test_table_args_none(self):
 
872
 
 
873
        class Foo2(Base):
 
874
 
 
875
            __tablename__ = 'foo'
 
876
            __table_args__ = None
 
877
            id = Column('id', Integer, primary_key=True)
 
878
 
 
879
        assert Foo2.__table__.kwargs == {}
 
880
 
 
881
    def test_table_args_dict_format(self):
 
882
 
 
883
        class Foo2(Base):
 
884
 
 
885
            __tablename__ = 'foo'
 
886
            __table_args__ = {'mysql_engine': 'InnoDB'}
 
887
            id = Column('id', Integer, primary_key=True)
 
888
 
 
889
        assert Foo2.__table__.kwargs['mysql_engine'] == 'InnoDB'
 
890
 
 
891
    def test_table_args_tuple_format(self):
 
892
        class Foo2(Base):
 
893
 
 
894
            __tablename__ = 'foo'
 
895
            __table_args__ = {'mysql_engine': 'InnoDB'}
 
896
            id = Column('id', Integer, primary_key=True)
 
897
 
 
898
        class Bar(Base):
 
899
 
 
900
            __tablename__ = 'bar'
 
901
            __table_args__ = ForeignKeyConstraint(['id'], ['foo.id']), \
 
902
                {'mysql_engine': 'InnoDB'}
 
903
            id = Column('id', Integer, primary_key=True)
 
904
 
 
905
        assert Bar.__table__.c.id.references(Foo2.__table__.c.id)
 
906
        assert Bar.__table__.kwargs['mysql_engine'] == 'InnoDB'
 
907
 
 
908
    def test_expression(self):
 
909
 
 
910
        class User(Base, fixtures.ComparableEntity):
 
911
 
 
912
            __tablename__ = 'users'
 
913
            id = Column('id', Integer, primary_key=True,
 
914
                        test_needs_autoincrement=True)
 
915
            name = Column('name', String(50))
 
916
            addresses = relationship('Address', backref='user')
 
917
 
 
918
        class Address(Base, fixtures.ComparableEntity):
 
919
 
 
920
            __tablename__ = 'addresses'
 
921
            id = Column('id', Integer, primary_key=True,
 
922
                        test_needs_autoincrement=True)
 
923
            email = Column('email', String(50))
 
924
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
925
 
 
926
        User.address_count = \
 
927
            sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
 
928
                    where(Address.user_id
 
929
                                   == User.id).as_scalar())
 
930
        Base.metadata.create_all()
 
931
        u1 = User(name='u1', addresses=[Address(email='one'),
 
932
                  Address(email='two')])
 
933
        sess = create_session()
 
934
        sess.add(u1)
 
935
        sess.flush()
 
936
        sess.expunge_all()
 
937
        eq_(sess.query(User).all(), [User(name='u1', address_count=2,
 
938
            addresses=[Address(email='one'), Address(email='two')])])
 
939
 
 
940
    def test_useless_declared_attr(self):
 
941
        class Address(Base, fixtures.ComparableEntity):
 
942
 
 
943
            __tablename__ = 'addresses'
 
944
            id = Column('id', Integer, primary_key=True,
 
945
                        test_needs_autoincrement=True)
 
946
            email = Column('email', String(50))
 
947
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
948
 
 
949
        class User(Base, fixtures.ComparableEntity):
 
950
 
 
951
            __tablename__ = 'users'
 
952
            id = Column('id', Integer, primary_key=True,
 
953
                        test_needs_autoincrement=True)
 
954
            name = Column('name', String(50))
 
955
            addresses = relationship('Address', backref='user')
 
956
 
 
957
            @declared_attr
 
958
            def address_count(cls):
 
959
                # this doesn't really gain us anything.  but if
 
960
                # one is used, lets have it function as expected...
 
961
                return sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
 
962
                        where(Address.user_id == cls.id))
 
963
 
 
964
        Base.metadata.create_all()
 
965
        u1 = User(name='u1', addresses=[Address(email='one'),
 
966
                  Address(email='two')])
 
967
        sess = create_session()
 
968
        sess.add(u1)
 
969
        sess.flush()
 
970
        sess.expunge_all()
 
971
        eq_(sess.query(User).all(), [User(name='u1', address_count=2,
 
972
            addresses=[Address(email='one'), Address(email='two')])])
 
973
 
 
974
    def test_declared_on_base_class(self):
 
975
        class MyBase(Base):
 
976
            __tablename__ = 'foo'
 
977
            id = Column(Integer, primary_key=True)
 
978
            @declared_attr
 
979
            def somecol(cls):
 
980
                return Column(Integer)
 
981
 
 
982
        class MyClass(MyBase):
 
983
            __tablename__ = 'bar'
 
984
            id = Column(Integer, ForeignKey('foo.id'), primary_key=True)
 
985
 
 
986
        # previously, the 'somecol' declared_attr would be ignored
 
987
        # by the mapping and would remain unused.  now we take
 
988
        # it as part of MyBase.
 
989
 
 
990
        assert 'somecol' in MyBase.__table__.c
 
991
        assert 'somecol' not in MyClass.__table__.c
 
992
 
 
993
    def test_column(self):
 
994
 
 
995
        class User(Base, fixtures.ComparableEntity):
 
996
 
 
997
            __tablename__ = 'users'
 
998
            id = Column('id', Integer, primary_key=True,
 
999
                        test_needs_autoincrement=True)
 
1000
            name = Column('name', String(50))
 
1001
 
 
1002
        User.a = Column('a', String(10))
 
1003
        User.b = Column(String(10))
 
1004
        Base.metadata.create_all()
 
1005
        u1 = User(name='u1', a='a', b='b')
 
1006
        eq_(u1.a, 'a')
 
1007
        eq_(User.a.get_history(u1), (['a'], (), ()))
 
1008
        sess = create_session()
 
1009
        sess.add(u1)
 
1010
        sess.flush()
 
1011
        sess.expunge_all()
 
1012
        eq_(sess.query(User).all(), [User(name='u1', a='a', b='b')])
 
1013
 
 
1014
    def test_column_properties(self):
 
1015
 
 
1016
        class Address(Base, fixtures.ComparableEntity):
 
1017
 
 
1018
            __tablename__ = 'addresses'
 
1019
            id = Column(Integer, primary_key=True,
 
1020
                        test_needs_autoincrement=True)
 
1021
            email = Column(String(50))
 
1022
            user_id = Column(Integer, ForeignKey('users.id'))
 
1023
 
 
1024
        class User(Base, fixtures.ComparableEntity):
 
1025
 
 
1026
            __tablename__ = 'users'
 
1027
            id = Column('id', Integer, primary_key=True,
 
1028
                        test_needs_autoincrement=True)
 
1029
            name = Column('name', String(50))
 
1030
 
 
1031
            adr_count = \
 
1032
                sa.orm.column_property(
 
1033
                    sa.select([sa.func.count(Address.id)],
 
1034
                        Address.user_id == id).as_scalar())
 
1035
            addresses = relationship(Address)
 
1036
 
 
1037
        Base.metadata.create_all()
 
1038
        u1 = User(name='u1', addresses=[Address(email='one'),
 
1039
                  Address(email='two')])
 
1040
        sess = create_session()
 
1041
        sess.add(u1)
 
1042
        sess.flush()
 
1043
        sess.expunge_all()
 
1044
        eq_(sess.query(User).all(), [User(name='u1', adr_count=2,
 
1045
            addresses=[Address(email='one'), Address(email='two')])])
 
1046
 
 
1047
    def test_column_properties_2(self):
 
1048
 
 
1049
        class Address(Base, fixtures.ComparableEntity):
 
1050
 
 
1051
            __tablename__ = 'addresses'
 
1052
            id = Column(Integer, primary_key=True)
 
1053
            email = Column(String(50))
 
1054
            user_id = Column(Integer, ForeignKey('users.id'))
 
1055
 
 
1056
        class User(Base, fixtures.ComparableEntity):
 
1057
 
 
1058
            __tablename__ = 'users'
 
1059
            id = Column('id', Integer, primary_key=True)
 
1060
            name = Column('name', String(50))
 
1061
 
 
1062
            # this is not "valid" but we want to test that Address.id
 
1063
            # doesnt get stuck into user's table
 
1064
 
 
1065
            adr_count = Address.id
 
1066
 
 
1067
        eq_(set(User.__table__.c.keys()), set(['id', 'name']))
 
1068
        eq_(set(Address.__table__.c.keys()), set(['id', 'email',
 
1069
            'user_id']))
 
1070
 
 
1071
    def test_deferred(self):
 
1072
 
 
1073
        class User(Base, fixtures.ComparableEntity):
 
1074
 
 
1075
            __tablename__ = 'users'
 
1076
            id = Column(Integer, primary_key=True,
 
1077
                        test_needs_autoincrement=True)
 
1078
            name = sa.orm.deferred(Column(String(50)))
 
1079
 
 
1080
        Base.metadata.create_all()
 
1081
        sess = create_session()
 
1082
        sess.add(User(name='u1'))
 
1083
        sess.flush()
 
1084
        sess.expunge_all()
 
1085
        u1 = sess.query(User).filter(User.name == 'u1').one()
 
1086
        assert 'name' not in u1.__dict__
 
1087
 
 
1088
        def go():
 
1089
            eq_(u1.name, 'u1')
 
1090
 
 
1091
        self.assert_sql_count(testing.db, go, 1)
 
1092
 
 
1093
    def test_composite_inline(self):
 
1094
        class AddressComposite(fixtures.ComparableEntity):
 
1095
            def __init__(self, street, state):
 
1096
                self.street = street
 
1097
                self.state = state
 
1098
            def __composite_values__(self):
 
1099
                return [self.street, self.state]
 
1100
 
 
1101
        class User(Base, fixtures.ComparableEntity):
 
1102
            __tablename__ = 'user'
 
1103
            id = Column(Integer, primary_key=True,
 
1104
                            test_needs_autoincrement=True)
 
1105
            address = composite(AddressComposite,
 
1106
                Column('street', String(50)),
 
1107
                Column('state', String(2)),
 
1108
            )
 
1109
 
 
1110
        Base.metadata.create_all()
 
1111
        sess = Session()
 
1112
        sess.add(User(
 
1113
                address=AddressComposite('123 anywhere street',
 
1114
                                'MD')
 
1115
                ))
 
1116
        sess.commit()
 
1117
        eq_(
 
1118
            sess.query(User).all(),
 
1119
            [User(address=AddressComposite('123 anywhere street',
 
1120
                                'MD'))]
 
1121
        )
 
1122
 
 
1123
    def test_composite_separate(self):
 
1124
        class AddressComposite(fixtures.ComparableEntity):
 
1125
            def __init__(self, street, state):
 
1126
                self.street = street
 
1127
                self.state = state
 
1128
            def __composite_values__(self):
 
1129
                return [self.street, self.state]
 
1130
 
 
1131
        class User(Base, fixtures.ComparableEntity):
 
1132
            __tablename__ = 'user'
 
1133
            id = Column(Integer, primary_key=True,
 
1134
                            test_needs_autoincrement=True)
 
1135
            street = Column(String(50))
 
1136
            state = Column(String(2))
 
1137
            address = composite(AddressComposite,
 
1138
                street, state)
 
1139
 
 
1140
        Base.metadata.create_all()
 
1141
        sess = Session()
 
1142
        sess.add(User(
 
1143
                address=AddressComposite('123 anywhere street',
 
1144
                                'MD')
 
1145
                ))
 
1146
        sess.commit()
 
1147
        eq_(
 
1148
            sess.query(User).all(),
 
1149
            [User(address=AddressComposite('123 anywhere street',
 
1150
                                'MD'))]
 
1151
        )
 
1152
 
 
1153
    def test_mapping_to_join(self):
 
1154
        users = Table('users', Base.metadata,
 
1155
            Column('id', Integer, primary_key=True)
 
1156
        )
 
1157
        addresses = Table('addresses', Base.metadata,
 
1158
            Column('id', Integer, primary_key=True),
 
1159
            Column('user_id', Integer, ForeignKey('users.id'))
 
1160
        )
 
1161
        usersaddresses = sa.join(users, addresses, users.c.id
 
1162
                                 == addresses.c.user_id)
 
1163
        class User(Base):
 
1164
            __table__ = usersaddresses
 
1165
            __table_args__ = {'primary_key':[users.c.id]}
 
1166
 
 
1167
            # need to use column_property for now
 
1168
            user_id = column_property(users.c.id, addresses.c.user_id)
 
1169
            address_id = addresses.c.id
 
1170
 
 
1171
        assert User.__mapper__.get_property('user_id').columns[0] \
 
1172
                                is users.c.id
 
1173
        assert User.__mapper__.get_property('user_id').columns[1] \
 
1174
                                is addresses.c.user_id
 
1175
 
 
1176
    def test_synonym_inline(self):
 
1177
 
 
1178
        class User(Base, fixtures.ComparableEntity):
 
1179
 
 
1180
            __tablename__ = 'users'
 
1181
            id = Column('id', Integer, primary_key=True,
 
1182
                        test_needs_autoincrement=True)
 
1183
            _name = Column('name', String(50))
 
1184
 
 
1185
            def _set_name(self, name):
 
1186
                self._name = 'SOMENAME ' + name
 
1187
 
 
1188
            def _get_name(self):
 
1189
                return self._name
 
1190
 
 
1191
            name = sa.orm.synonym('_name',
 
1192
                                  descriptor=property(_get_name,
 
1193
                                  _set_name))
 
1194
 
 
1195
        Base.metadata.create_all()
 
1196
        sess = create_session()
 
1197
        u1 = User(name='someuser')
 
1198
        eq_(u1.name, 'SOMENAME someuser')
 
1199
        sess.add(u1)
 
1200
        sess.flush()
 
1201
        eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
 
1202
            ).one(), u1)
 
1203
 
 
1204
    def test_synonym_no_descriptor(self):
 
1205
        from sqlalchemy.orm.properties import ColumnProperty
 
1206
 
 
1207
        class CustomCompare(ColumnProperty.Comparator):
 
1208
 
 
1209
            __hash__ = None
 
1210
 
 
1211
            def __eq__(self, other):
 
1212
                return self.__clause_element__() == other + ' FOO'
 
1213
 
 
1214
        class User(Base, fixtures.ComparableEntity):
 
1215
 
 
1216
            __tablename__ = 'users'
 
1217
            id = Column('id', Integer, primary_key=True,
 
1218
                        test_needs_autoincrement=True)
 
1219
            _name = Column('name', String(50))
 
1220
            name = sa.orm.synonym('_name',
 
1221
                                  comparator_factory=CustomCompare)
 
1222
 
 
1223
        Base.metadata.create_all()
 
1224
        sess = create_session()
 
1225
        u1 = User(name='someuser FOO')
 
1226
        sess.add(u1)
 
1227
        sess.flush()
 
1228
        eq_(sess.query(User).filter(User.name == 'someuser').one(), u1)
 
1229
 
 
1230
    def test_synonym_added(self):
 
1231
 
 
1232
        class User(Base, fixtures.ComparableEntity):
 
1233
 
 
1234
            __tablename__ = 'users'
 
1235
            id = Column('id', Integer, primary_key=True,
 
1236
                        test_needs_autoincrement=True)
 
1237
            _name = Column('name', String(50))
 
1238
 
 
1239
            def _set_name(self, name):
 
1240
                self._name = 'SOMENAME ' + name
 
1241
 
 
1242
            def _get_name(self):
 
1243
                return self._name
 
1244
 
 
1245
            name = property(_get_name, _set_name)
 
1246
 
 
1247
        User.name = sa.orm.synonym('_name', descriptor=User.name)
 
1248
        Base.metadata.create_all()
 
1249
        sess = create_session()
 
1250
        u1 = User(name='someuser')
 
1251
        eq_(u1.name, 'SOMENAME someuser')
 
1252
        sess.add(u1)
 
1253
        sess.flush()
 
1254
        eq_(sess.query(User).filter(User.name == 'SOMENAME someuser'
 
1255
            ).one(), u1)
 
1256
 
 
1257
    def test_reentrant_compile_via_foreignkey(self):
 
1258
 
 
1259
        class User(Base, fixtures.ComparableEntity):
 
1260
 
 
1261
            __tablename__ = 'users'
 
1262
            id = Column('id', Integer, primary_key=True,
 
1263
                        test_needs_autoincrement=True)
 
1264
            name = Column('name', String(50))
 
1265
            addresses = relationship('Address', backref='user')
 
1266
 
 
1267
        class Address(Base, fixtures.ComparableEntity):
 
1268
 
 
1269
            __tablename__ = 'addresses'
 
1270
            id = Column('id', Integer, primary_key=True,
 
1271
                        test_needs_autoincrement=True)
 
1272
            email = Column('email', String(50))
 
1273
            user_id = Column('user_id', Integer, ForeignKey(User.id))
 
1274
 
 
1275
        # previous versions would force a re-entrant mapper compile via
 
1276
        # the User.id inside the ForeignKey but this is no longer the
 
1277
        # case
 
1278
 
 
1279
        sa.orm.configure_mappers()
 
1280
        eq_(str(list(Address.user_id.property.columns[0].foreign_keys)[0]),
 
1281
            "ForeignKey('users.id')")
 
1282
        Base.metadata.create_all()
 
1283
        u1 = User(name='u1', addresses=[Address(email='one'),
 
1284
                  Address(email='two')])
 
1285
        sess = create_session()
 
1286
        sess.add(u1)
 
1287
        sess.flush()
 
1288
        sess.expunge_all()
 
1289
        eq_(sess.query(User).all(), [User(name='u1',
 
1290
            addresses=[Address(email='one'), Address(email='two')])])
 
1291
 
 
1292
    def test_relationship_reference(self):
 
1293
 
 
1294
        class Address(Base, fixtures.ComparableEntity):
 
1295
 
 
1296
            __tablename__ = 'addresses'
 
1297
            id = Column('id', Integer, primary_key=True,
 
1298
                        test_needs_autoincrement=True)
 
1299
            email = Column('email', String(50))
 
1300
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
1301
 
 
1302
        class User(Base, fixtures.ComparableEntity):
 
1303
 
 
1304
            __tablename__ = 'users'
 
1305
            id = Column('id', Integer, primary_key=True,
 
1306
                        test_needs_autoincrement=True)
 
1307
            name = Column('name', String(50))
 
1308
            addresses = relationship('Address', backref='user',
 
1309
                    primaryjoin=id == Address.user_id)
 
1310
 
 
1311
        User.address_count = \
 
1312
            sa.orm.column_property(sa.select([sa.func.count(Address.id)]).
 
1313
                    where(Address.user_id
 
1314
                                   == User.id).as_scalar())
 
1315
        Base.metadata.create_all()
 
1316
        u1 = User(name='u1', addresses=[Address(email='one'),
 
1317
                  Address(email='two')])
 
1318
        sess = create_session()
 
1319
        sess.add(u1)
 
1320
        sess.flush()
 
1321
        sess.expunge_all()
 
1322
        eq_(sess.query(User).all(), [User(name='u1', address_count=2,
 
1323
            addresses=[Address(email='one'), Address(email='two')])])
 
1324
 
 
1325
    def test_pk_with_fk_init(self):
 
1326
 
 
1327
        class Bar(Base):
 
1328
 
 
1329
            __tablename__ = 'bar'
 
1330
            id = sa.Column(sa.Integer, sa.ForeignKey('foo.id'),
 
1331
                           primary_key=True)
 
1332
            ex = sa.Column(sa.Integer, primary_key=True)
 
1333
 
 
1334
        class Foo(Base):
 
1335
 
 
1336
            __tablename__ = 'foo'
 
1337
            id = sa.Column(sa.Integer, primary_key=True)
 
1338
            bars = sa.orm.relationship(Bar)
 
1339
 
 
1340
        assert Bar.__mapper__.primary_key[0] is Bar.__table__.c.id
 
1341
        assert Bar.__mapper__.primary_key[1] is Bar.__table__.c.ex
 
1342
 
 
1343
    def test_with_explicit_autoloaded(self):
 
1344
        meta = MetaData(testing.db)
 
1345
        t1 = Table('t1', meta, Column('id', String(50),
 
1346
                   primary_key=True, test_needs_autoincrement=True),
 
1347
                   Column('data', String(50)))
 
1348
        meta.create_all()
 
1349
        try:
 
1350
 
 
1351
            class MyObj(Base):
 
1352
 
 
1353
                __table__ = Table('t1', Base.metadata, autoload=True)
 
1354
 
 
1355
            sess = create_session()
 
1356
            m = MyObj(id='someid', data='somedata')
 
1357
            sess.add(m)
 
1358
            sess.flush()
 
1359
            eq_(t1.select().execute().fetchall(), [('someid', 'somedata'
 
1360
                )])
 
1361
        finally:
 
1362
            meta.drop_all()
 
1363
 
 
1364
    def test_synonym_for(self):
 
1365
 
 
1366
        class User(Base, fixtures.ComparableEntity):
 
1367
 
 
1368
            __tablename__ = 'users'
 
1369
            id = Column('id', Integer, primary_key=True,
 
1370
                        test_needs_autoincrement=True)
 
1371
            name = Column('name', String(50))
 
1372
 
 
1373
            @decl.synonym_for('name')
 
1374
            @property
 
1375
            def namesyn(self):
 
1376
                return self.name
 
1377
 
 
1378
        Base.metadata.create_all()
 
1379
        sess = create_session()
 
1380
        u1 = User(name='someuser')
 
1381
        eq_(u1.name, 'someuser')
 
1382
        eq_(u1.namesyn, 'someuser')
 
1383
        sess.add(u1)
 
1384
        sess.flush()
 
1385
        rt = sess.query(User).filter(User.namesyn == 'someuser').one()
 
1386
        eq_(rt, u1)
 
1387
 
 
1388
    def test_comparable_using(self):
 
1389
 
 
1390
        class NameComparator(sa.orm.PropComparator):
 
1391
 
 
1392
            @property
 
1393
            def upperself(self):
 
1394
                cls = self.prop.parent.class_
 
1395
                col = getattr(cls, 'name')
 
1396
                return sa.func.upper(col)
 
1397
 
 
1398
            def operate(
 
1399
                self,
 
1400
                op,
 
1401
                other,
 
1402
                **kw
 
1403
                ):
 
1404
                return op(self.upperself, other, **kw)
 
1405
 
 
1406
        class User(Base, fixtures.ComparableEntity):
 
1407
 
 
1408
            __tablename__ = 'users'
 
1409
            id = Column('id', Integer, primary_key=True,
 
1410
                        test_needs_autoincrement=True)
 
1411
            name = Column('name', String(50))
 
1412
 
 
1413
            @decl.comparable_using(NameComparator)
 
1414
            @property
 
1415
            def uc_name(self):
 
1416
                return self.name is not None and self.name.upper() \
 
1417
                    or None
 
1418
 
 
1419
        Base.metadata.create_all()
 
1420
        sess = create_session()
 
1421
        u1 = User(name='someuser')
 
1422
        eq_(u1.name, 'someuser', u1.name)
 
1423
        eq_(u1.uc_name, 'SOMEUSER', u1.uc_name)
 
1424
        sess.add(u1)
 
1425
        sess.flush()
 
1426
        sess.expunge_all()
 
1427
        rt = sess.query(User).filter(User.uc_name == 'SOMEUSER').one()
 
1428
        eq_(rt, u1)
 
1429
        sess.expunge_all()
 
1430
        rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE'
 
1431
                )).one()
 
1432
        eq_(rt, u1)
 
1433
 
 
1434
    def test_duplicate_classes_in_base(self):
 
1435
 
 
1436
        class Test(Base):
 
1437
            __tablename__ = 'a'
 
1438
            id = Column(Integer, primary_key=True)
 
1439
 
 
1440
        assert_raises_message(
 
1441
            sa.exc.SAWarning,
 
1442
            "This declarative base already contains a class with ",
 
1443
            lambda: type(Base)("Test", (Base,), dict(
 
1444
                __tablename__='b',
 
1445
                id=Column(Integer, primary_key=True)
 
1446
            ))
 
1447
        )
 
1448
 
 
1449
 
 
1450
 
 
1451
def _produce_test(inline, stringbased):
 
1452
 
 
1453
    class ExplicitJoinTest(fixtures.MappedTest):
 
1454
 
 
1455
        @classmethod
 
1456
        def define_tables(cls, metadata):
 
1457
            global User, Address
 
1458
            Base = decl.declarative_base(metadata=metadata)
 
1459
 
 
1460
            class User(Base, fixtures.ComparableEntity):
 
1461
 
 
1462
                __tablename__ = 'users'
 
1463
                id = Column(Integer, primary_key=True,
 
1464
                            test_needs_autoincrement=True)
 
1465
                name = Column(String(50))
 
1466
 
 
1467
            class Address(Base, fixtures.ComparableEntity):
 
1468
 
 
1469
                __tablename__ = 'addresses'
 
1470
                id = Column(Integer, primary_key=True,
 
1471
                            test_needs_autoincrement=True)
 
1472
                email = Column(String(50))
 
1473
                user_id = Column(Integer, ForeignKey('users.id'))
 
1474
                if inline:
 
1475
                    if stringbased:
 
1476
                        user = relationship('User',
 
1477
                                primaryjoin='User.id==Address.user_id',
 
1478
                                backref='addresses')
 
1479
                    else:
 
1480
                        user = relationship(User, primaryjoin=User.id
 
1481
                                == user_id, backref='addresses')
 
1482
 
 
1483
            if not inline:
 
1484
                configure_mappers()
 
1485
                if stringbased:
 
1486
                    Address.user = relationship('User',
 
1487
                            primaryjoin='User.id==Address.user_id',
 
1488
                            backref='addresses')
 
1489
                else:
 
1490
                    Address.user = relationship(User,
 
1491
                            primaryjoin=User.id == Address.user_id,
 
1492
                            backref='addresses')
 
1493
 
 
1494
        @classmethod
 
1495
        def insert_data(cls):
 
1496
            params = [dict(zip(('id', 'name'), column_values))
 
1497
                      for column_values in [(7, 'jack'), (8, 'ed'), (9,
 
1498
                      'fred'), (10, 'chuck')]]
 
1499
            User.__table__.insert().execute(params)
 
1500
            Address.__table__.insert().execute([dict(zip(('id',
 
1501
                    'user_id', 'email'), column_values))
 
1502
                    for column_values in [(1, 7, 'jack@bean.com'), (2,
 
1503
                    8, 'ed@wood.com'), (3, 8, 'ed@bettyboop.com'), (4,
 
1504
                    8, 'ed@lala.com'), (5, 9, 'fred@fred.com')]])
 
1505
 
 
1506
        def test_aliased_join(self):
 
1507
 
 
1508
            # this query will screw up if the aliasing enabled in
 
1509
            # query.join() gets applied to the right half of the join
 
1510
            # condition inside the any(). the join condition inside of
 
1511
            # any() comes from the "primaryjoin" of the relationship,
 
1512
            # and should not be annotated with _orm_adapt.
 
1513
            # PropertyLoader.Comparator will annotate the left side with
 
1514
            # _orm_adapt, though.
 
1515
 
 
1516
            sess = create_session()
 
1517
            eq_(sess.query(User).join(User.addresses,
 
1518
                aliased=True).filter(Address.email == 'ed@wood.com'
 
1519
                ).filter(User.addresses.any(Address.email
 
1520
                == 'jack@bean.com')).all(), [])
 
1521
 
 
1522
    ExplicitJoinTest.__name__ = 'ExplicitJoinTest%s%s' % (inline
 
1523
            and 'Inline' or 'Separate', stringbased and 'String'
 
1524
            or 'Literal')
 
1525
    return ExplicitJoinTest
 
1526
 
 
1527
for inline in True, False:
 
1528
    for stringbased in True, False:
 
1529
        testclass = _produce_test(inline, stringbased)
 
1530
        exec '%s = testclass' % testclass.__name__
 
1531
        del testclass
 
1532