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