1 2from sqlalchemy.testing import eq_, assert_raises, \ 3 assert_raises_message, is_, is_true, is_false 4from sqlalchemy.ext import declarative as decl 5import sqlalchemy as sa 6from sqlalchemy import testing 7from sqlalchemy import Integer, String, ForeignKey 8from sqlalchemy.testing.schema import Table, Column 9from sqlalchemy.orm import relationship, create_session, class_mapper, \ 10 configure_mappers, clear_mappers, \ 11 polymorphic_union, deferred, Session 12from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \ 13 ConcreteBase, has_inherited_table 14from sqlalchemy.testing import fixtures, mock 15from test.orm.test_events import _RemoveListeners 16 17Base = None 18 19 20class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults): 21 22 def setup(self): 23 global Base 24 Base = decl.declarative_base(testing.db) 25 26 def teardown(self): 27 Session.close_all() 28 clear_mappers() 29 Base.metadata.drop_all() 30 31 32class DeclarativeInheritanceTest(DeclarativeTestBase): 33 34 def test_we_must_copy_mapper_args(self): 35 36 class Person(Base): 37 38 __tablename__ = 'people' 39 id = Column(Integer, primary_key=True) 40 discriminator = Column('type', String(50)) 41 __mapper_args__ = {'polymorphic_on': discriminator, 42 'polymorphic_identity': 'person'} 43 44 class Engineer(Person): 45 46 primary_language = Column(String(50)) 47 48 assert 'inherits' not in Person.__mapper_args__ 49 assert class_mapper(Engineer).polymorphic_identity is None 50 assert class_mapper(Engineer).polymorphic_on is Person.__table__.c.type 51 52 def test_we_must_only_copy_column_mapper_args(self): 53 54 class Person(Base): 55 56 __tablename__ = 'people' 57 id = Column(Integer, primary_key=True) 58 a = Column(Integer) 59 b = Column(Integer) 60 c = Column(Integer) 61 d = Column(Integer) 62 discriminator = Column('type', String(50)) 63 __mapper_args__ = {'polymorphic_on': discriminator, 64 'polymorphic_identity': 'person', 65 'version_id_col': 'a', 66 'column_prefix': 'bar', 67 'include_properties': ['id', 'a', 'b'], 68 } 69 assert class_mapper(Person).version_id_col == 'a' 70 assert class_mapper(Person).include_properties == set(['id', 'a', 'b']) 71 72 def test_custom_join_condition(self): 73 74 class Foo(Base): 75 76 __tablename__ = 'foo' 77 id = Column('id', Integer, primary_key=True) 78 79 class Bar(Foo): 80 81 __tablename__ = 'bar' 82 bar_id = Column('id', Integer, primary_key=True) 83 foo_id = Column('foo_id', Integer) 84 __mapper_args__ = {'inherit_condition': foo_id == Foo.id} 85 86 # compile succeeds because inherit_condition is honored 87 88 configure_mappers() 89 90 def test_joined(self): 91 92 class Company(Base, fixtures.ComparableEntity): 93 94 __tablename__ = 'companies' 95 id = Column('id', Integer, primary_key=True, 96 test_needs_autoincrement=True) 97 name = Column('name', String(50)) 98 employees = relationship('Person') 99 100 class Person(Base, fixtures.ComparableEntity): 101 102 __tablename__ = 'people' 103 id = Column('id', Integer, primary_key=True, 104 test_needs_autoincrement=True) 105 company_id = Column('company_id', Integer, 106 ForeignKey('companies.id')) 107 name = Column('name', String(50)) 108 discriminator = Column('type', String(50)) 109 __mapper_args__ = {'polymorphic_on': discriminator} 110 111 class Engineer(Person): 112 113 __tablename__ = 'engineers' 114 __mapper_args__ = {'polymorphic_identity': 'engineer'} 115 id = Column('id', Integer, ForeignKey('people.id'), 116 primary_key=True) 117 primary_language = Column('primary_language', String(50)) 118 119 class Manager(Person): 120 121 __tablename__ = 'managers' 122 __mapper_args__ = {'polymorphic_identity': 'manager'} 123 id = Column('id', Integer, ForeignKey('people.id'), 124 primary_key=True) 125 golf_swing = Column('golf_swing', String(50)) 126 127 Base.metadata.create_all() 128 sess = create_session() 129 c1 = Company( 130 name='MegaCorp, Inc.', 131 employees=[ 132 Engineer(name='dilbert', primary_language='java'), 133 Engineer(name='wally', primary_language='c++'), 134 Manager(name='dogbert', golf_swing='fore!')]) 135 136 c2 = Company(name='Elbonia, Inc.', 137 employees=[Engineer(name='vlad', 138 primary_language='cobol')]) 139 sess.add(c1) 140 sess.add(c2) 141 sess.flush() 142 sess.expunge_all() 143 eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). 144 any(Engineer.primary_language 145 == 'cobol')).first(), c2) 146 147 # ensure that the Manager mapper was compiled with the Manager id 148 # column as higher priority. this ensures that "Manager.id" 149 # is appropriately treated as the "id" column in the "manager" 150 # table (reversed from 0.6's behavior.) 151 152 eq_( 153 Manager.id.property.columns, 154 [Manager.__table__.c.id, Person.__table__.c.id] 155 ) 156 157 # assert that the "id" column is available without a second 158 # load. as of 0.7, the ColumnProperty tests all columns 159 # in its list to see which is present in the row. 160 161 sess.expunge_all() 162 163 def go(): 164 assert sess.query(Manager).filter(Manager.name == 'dogbert' 165 ).one().id 166 self.assert_sql_count(testing.db, go, 1) 167 sess.expunge_all() 168 169 def go(): 170 assert sess.query(Person).filter(Manager.name == 'dogbert' 171 ).one().id 172 173 self.assert_sql_count(testing.db, go, 1) 174 175 def test_add_subcol_after_the_fact(self): 176 177 class Person(Base, fixtures.ComparableEntity): 178 179 __tablename__ = 'people' 180 id = Column('id', Integer, primary_key=True, 181 test_needs_autoincrement=True) 182 name = Column('name', String(50)) 183 discriminator = Column('type', String(50)) 184 __mapper_args__ = {'polymorphic_on': discriminator} 185 186 class Engineer(Person): 187 188 __tablename__ = 'engineers' 189 __mapper_args__ = {'polymorphic_identity': 'engineer'} 190 id = Column('id', Integer, ForeignKey('people.id'), 191 primary_key=True) 192 193 Engineer.primary_language = Column('primary_language', 194 String(50)) 195 Base.metadata.create_all() 196 sess = create_session() 197 e1 = Engineer(primary_language='java', name='dilbert') 198 sess.add(e1) 199 sess.flush() 200 sess.expunge_all() 201 eq_(sess.query(Person).first(), 202 Engineer(primary_language='java', name='dilbert')) 203 204 def test_add_parentcol_after_the_fact(self): 205 206 class Person(Base, fixtures.ComparableEntity): 207 208 __tablename__ = 'people' 209 id = Column('id', Integer, primary_key=True, 210 test_needs_autoincrement=True) 211 discriminator = Column('type', String(50)) 212 __mapper_args__ = {'polymorphic_on': discriminator} 213 214 class Engineer(Person): 215 216 __tablename__ = 'engineers' 217 __mapper_args__ = {'polymorphic_identity': 'engineer'} 218 primary_language = Column(String(50)) 219 id = Column('id', Integer, ForeignKey('people.id'), 220 primary_key=True) 221 222 Person.name = Column('name', String(50)) 223 Base.metadata.create_all() 224 sess = create_session() 225 e1 = Engineer(primary_language='java', name='dilbert') 226 sess.add(e1) 227 sess.flush() 228 sess.expunge_all() 229 eq_(sess.query(Person).first(), 230 Engineer(primary_language='java', name='dilbert')) 231 232 def test_add_sub_parentcol_after_the_fact(self): 233 234 class Person(Base, fixtures.ComparableEntity): 235 236 __tablename__ = 'people' 237 id = Column('id', Integer, primary_key=True, 238 test_needs_autoincrement=True) 239 discriminator = Column('type', String(50)) 240 __mapper_args__ = {'polymorphic_on': discriminator} 241 242 class Engineer(Person): 243 244 __tablename__ = 'engineers' 245 __mapper_args__ = {'polymorphic_identity': 'engineer'} 246 primary_language = Column(String(50)) 247 id = Column('id', Integer, ForeignKey('people.id'), 248 primary_key=True) 249 250 class Admin(Engineer): 251 252 __tablename__ = 'admins' 253 __mapper_args__ = {'polymorphic_identity': 'admin'} 254 workstation = Column(String(50)) 255 id = Column('id', Integer, ForeignKey('engineers.id'), 256 primary_key=True) 257 258 Person.name = Column('name', String(50)) 259 Base.metadata.create_all() 260 sess = create_session() 261 e1 = Admin(primary_language='java', name='dilbert', 262 workstation='foo') 263 sess.add(e1) 264 sess.flush() 265 sess.expunge_all() 266 eq_(sess.query(Person).first(), 267 Admin(primary_language='java', name='dilbert', workstation='foo')) 268 269 def test_subclass_mixin(self): 270 271 class Person(Base, fixtures.ComparableEntity): 272 273 __tablename__ = 'people' 274 id = Column('id', Integer, primary_key=True) 275 name = Column('name', String(50)) 276 discriminator = Column('type', String(50)) 277 __mapper_args__ = {'polymorphic_on': discriminator} 278 279 class MyMixin(object): 280 281 pass 282 283 class Engineer(MyMixin, Person): 284 285 __tablename__ = 'engineers' 286 __mapper_args__ = {'polymorphic_identity': 'engineer'} 287 id = Column('id', Integer, ForeignKey('people.id'), 288 primary_key=True) 289 primary_language = Column('primary_language', String(50)) 290 291 assert class_mapper(Engineer).inherits is class_mapper(Person) 292 293 def test_with_undefined_foreignkey(self): 294 295 class Parent(Base): 296 297 __tablename__ = 'parent' 298 id = Column('id', Integer, primary_key=True) 299 tp = Column('type', String(50)) 300 __mapper_args__ = dict(polymorphic_on=tp) 301 302 class Child1(Parent): 303 304 __tablename__ = 'child1' 305 id = Column('id', Integer, ForeignKey('parent.id'), 306 primary_key=True) 307 related_child2 = Column('c2', Integer, 308 ForeignKey('child2.id')) 309 __mapper_args__ = dict(polymorphic_identity='child1') 310 311 # no exception is raised by the ForeignKey to "child2" even 312 # though child2 doesn't exist yet 313 314 class Child2(Parent): 315 316 __tablename__ = 'child2' 317 id = Column('id', Integer, ForeignKey('parent.id'), 318 primary_key=True) 319 related_child1 = Column('c1', Integer) 320 __mapper_args__ = dict(polymorphic_identity='child2') 321 322 sa.orm.configure_mappers() # no exceptions here 323 324 def test_foreign_keys_with_col(self): 325 """Test that foreign keys that reference a literal 'id' subclass 326 'id' attribute behave intuitively. 327 328 See [ticket:1892]. 329 330 """ 331 332 class Booking(Base): 333 __tablename__ = 'booking' 334 id = Column(Integer, primary_key=True) 335 336 class PlanBooking(Booking): 337 __tablename__ = 'plan_booking' 338 id = Column(Integer, ForeignKey(Booking.id), 339 primary_key=True) 340 341 # referencing PlanBooking.id gives us the column 342 # on plan_booking, not booking 343 class FeatureBooking(Booking): 344 __tablename__ = 'feature_booking' 345 id = Column(Integer, ForeignKey(Booking.id), 346 primary_key=True) 347 plan_booking_id = Column(Integer, 348 ForeignKey(PlanBooking.id)) 349 350 plan_booking = relationship(PlanBooking, 351 backref='feature_bookings') 352 353 assert FeatureBooking.__table__.c.plan_booking_id.\ 354 references(PlanBooking.__table__.c.id) 355 356 assert FeatureBooking.__table__.c.id.\ 357 references(Booking.__table__.c.id) 358 359 def test_single_colsonbase(self): 360 """test single inheritance where all the columns are on the base 361 class.""" 362 363 class Company(Base, fixtures.ComparableEntity): 364 365 __tablename__ = 'companies' 366 id = Column('id', Integer, primary_key=True, 367 test_needs_autoincrement=True) 368 name = Column('name', String(50)) 369 employees = relationship('Person') 370 371 class Person(Base, fixtures.ComparableEntity): 372 373 __tablename__ = 'people' 374 id = Column('id', Integer, primary_key=True, 375 test_needs_autoincrement=True) 376 company_id = Column('company_id', Integer, 377 ForeignKey('companies.id')) 378 name = Column('name', String(50)) 379 discriminator = Column('type', String(50)) 380 primary_language = Column('primary_language', String(50)) 381 golf_swing = Column('golf_swing', String(50)) 382 __mapper_args__ = {'polymorphic_on': discriminator} 383 384 class Engineer(Person): 385 386 __mapper_args__ = {'polymorphic_identity': 'engineer'} 387 388 class Manager(Person): 389 390 __mapper_args__ = {'polymorphic_identity': 'manager'} 391 392 Base.metadata.create_all() 393 sess = create_session() 394 c1 = Company( 395 name='MegaCorp, Inc.', 396 employees=[ 397 Engineer(name='dilbert', primary_language='java'), 398 Engineer(name='wally', primary_language='c++'), 399 Manager(name='dogbert', golf_swing='fore!')]) 400 401 c2 = Company(name='Elbonia, Inc.', 402 employees=[Engineer(name='vlad', 403 primary_language='cobol')]) 404 sess.add(c1) 405 sess.add(c2) 406 sess.flush() 407 sess.expunge_all() 408 eq_(sess.query(Person).filter(Engineer.primary_language 409 == 'cobol').first(), 410 Engineer(name='vlad')) 411 eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). 412 any(Engineer.primary_language 413 == 'cobol')).first(), c2) 414 415 def test_single_colsonsub(self): 416 """test single inheritance where the columns are local to their 417 class. 418 419 this is a newer usage. 420 421 """ 422 423 class Company(Base, fixtures.ComparableEntity): 424 425 __tablename__ = 'companies' 426 id = Column('id', Integer, primary_key=True, 427 test_needs_autoincrement=True) 428 name = Column('name', String(50)) 429 employees = relationship('Person') 430 431 class Person(Base, fixtures.ComparableEntity): 432 433 __tablename__ = 'people' 434 id = Column(Integer, primary_key=True, 435 test_needs_autoincrement=True) 436 company_id = Column(Integer, ForeignKey('companies.id')) 437 name = Column(String(50)) 438 discriminator = Column('type', String(50)) 439 __mapper_args__ = {'polymorphic_on': discriminator} 440 441 class Engineer(Person): 442 443 __mapper_args__ = {'polymorphic_identity': 'engineer'} 444 primary_language = Column(String(50)) 445 446 class Manager(Person): 447 448 __mapper_args__ = {'polymorphic_identity': 'manager'} 449 golf_swing = Column(String(50)) 450 451 # we have here a situation that is somewhat unique. the Person 452 # class is mapped to the "people" table, but it was mapped when 453 # the table did not include the "primary_language" or 454 # "golf_swing" columns. declarative will also manipulate the 455 # exclude_properties collection so that sibling classes don't 456 # cross-pollinate. 457 458 assert Person.__table__.c.company_id is not None 459 assert Person.__table__.c.golf_swing is not None 460 assert Person.__table__.c.primary_language is not None 461 assert Engineer.primary_language is not None 462 assert Manager.golf_swing is not None 463 assert not hasattr(Person, 'primary_language') 464 assert not hasattr(Person, 'golf_swing') 465 assert not hasattr(Engineer, 'golf_swing') 466 assert not hasattr(Manager, 'primary_language') 467 Base.metadata.create_all() 468 sess = create_session() 469 e1 = Engineer(name='dilbert', primary_language='java') 470 e2 = Engineer(name='wally', primary_language='c++') 471 m1 = Manager(name='dogbert', golf_swing='fore!') 472 c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1]) 473 e3 = Engineer(name='vlad', primary_language='cobol') 474 c2 = Company(name='Elbonia, Inc.', employees=[e3]) 475 sess.add(c1) 476 sess.add(c2) 477 sess.flush() 478 sess.expunge_all() 479 eq_(sess.query(Person).filter(Engineer.primary_language 480 == 'cobol').first(), 481 Engineer(name='vlad')) 482 eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). 483 any(Engineer.primary_language 484 == 'cobol')).first(), c2) 485 eq_(sess.query(Engineer).filter_by(primary_language='cobol' 486 ).one(), 487 Engineer(name='vlad', primary_language='cobol')) 488 489 def test_single_cols_on_sub_base_of_joined(self): 490 """test [ticket:3895]""" 491 492 class Person(Base): 493 __tablename__ = "person" 494 495 id = Column(Integer, primary_key=True) 496 type = Column(String) 497 498 __mapper_args__ = { 499 "polymorphic_on": type, 500 } 501 502 class Contractor(Person): 503 contractor_field = Column(String) 504 505 __mapper_args__ = { 506 "polymorphic_identity": "contractor", 507 } 508 509 class Employee(Person): 510 __tablename__ = "employee" 511 512 id = Column(Integer, ForeignKey(Person.id), primary_key=True) 513 514 class Engineer(Employee): 515 __mapper_args__ = { 516 "polymorphic_identity": "engineer", 517 } 518 519 configure_mappers() 520 521 is_false(hasattr(Person, 'contractor_field')) 522 is_true(hasattr(Contractor, 'contractor_field')) 523 is_false(hasattr(Employee, 'contractor_field')) 524 is_false(hasattr(Engineer, 'contractor_field')) 525 526 def test_single_cols_on_sub_to_joined(self): 527 """test [ticket:3797]""" 528 529 class BaseUser(Base): 530 __tablename__ = 'root' 531 532 id = Column(Integer, primary_key=True) 533 row_type = Column(String) 534 535 __mapper_args__ = { 536 'polymorphic_on': row_type, 537 'polymorphic_identity': 'baseuser' 538 } 539 540 class User(BaseUser): 541 __tablename__ = 'user' 542 543 __mapper_args__ = { 544 'polymorphic_identity': 'user' 545 } 546 547 baseuser_id = Column( 548 Integer, ForeignKey('root.id'), primary_key=True) 549 550 class Bat(Base): 551 __tablename__ = 'bat' 552 id = Column(Integer, primary_key=True) 553 554 class Thing(Base): 555 __tablename__ = 'thing' 556 557 id = Column(Integer, primary_key=True) 558 559 owner_id = Column(Integer, ForeignKey('user.baseuser_id')) 560 owner = relationship('User') 561 562 class SubUser(User): 563 __mapper_args__ = { 564 'polymorphic_identity': 'subuser' 565 } 566 567 sub_user_custom_thing = Column(Integer, ForeignKey('bat.id')) 568 569 eq_( 570 User.__table__.foreign_keys, 571 User.baseuser_id.foreign_keys.union( 572 SubUser.sub_user_custom_thing.foreign_keys)) 573 is_true(Thing.owner.property.primaryjoin.compare( 574 Thing.owner_id == User.baseuser_id)) 575 576 def test_single_constraint_on_sub(self): 577 """test the somewhat unusual case of [ticket:3341]""" 578 579 class Person(Base, fixtures.ComparableEntity): 580 581 __tablename__ = 'people' 582 id = Column(Integer, primary_key=True, 583 test_needs_autoincrement=True) 584 name = Column(String(50)) 585 discriminator = Column('type', String(50)) 586 __mapper_args__ = {'polymorphic_on': discriminator} 587 588 class Engineer(Person): 589 590 __mapper_args__ = {'polymorphic_identity': 'engineer'} 591 primary_language = Column(String(50)) 592 593 __hack_args_one__ = sa.UniqueConstraint( 594 Person.name, primary_language) 595 __hack_args_two__ = sa.CheckConstraint( 596 Person.name != primary_language) 597 598 uq = [c for c in Person.__table__.constraints 599 if isinstance(c, sa.UniqueConstraint)][0] 600 ck = [c for c in Person.__table__.constraints 601 if isinstance(c, sa.CheckConstraint)][0] 602 eq_( 603 list(uq.columns), 604 [Person.__table__.c.name, Person.__table__.c.primary_language] 605 ) 606 eq_( 607 list(ck.columns), 608 [Person.__table__.c.name, Person.__table__.c.primary_language] 609 ) 610 611 @testing.skip_if(lambda: testing.against('oracle'), 612 "Test has an empty insert in it at the moment") 613 def test_columns_single_inheritance_conflict_resolution(self): 614 """Test that a declared_attr can return the existing column and it will 615 be ignored. this allows conditional columns to be added. 616 617 See [ticket:2472]. 618 619 """ 620 class Person(Base): 621 __tablename__ = 'person' 622 id = Column(Integer, primary_key=True) 623 624 class Engineer(Person): 625 626 """single table inheritance""" 627 628 @declared_attr 629 def target_id(cls): 630 return cls.__table__.c.get( 631 'target_id', 632 Column(Integer, ForeignKey('other.id'))) 633 634 @declared_attr 635 def target(cls): 636 return relationship("Other") 637 638 class Manager(Person): 639 640 """single table inheritance""" 641 642 @declared_attr 643 def target_id(cls): 644 return cls.__table__.c.get( 645 'target_id', 646 Column(Integer, ForeignKey('other.id'))) 647 648 @declared_attr 649 def target(cls): 650 return relationship("Other") 651 652 class Other(Base): 653 __tablename__ = 'other' 654 id = Column(Integer, primary_key=True) 655 656 is_( 657 Engineer.target_id.property.columns[0], 658 Person.__table__.c.target_id 659 ) 660 is_( 661 Manager.target_id.property.columns[0], 662 Person.__table__.c.target_id 663 ) 664 # do a brief round trip on this 665 Base.metadata.create_all() 666 session = Session() 667 o1, o2 = Other(), Other() 668 session.add_all([ 669 Engineer(target=o1), 670 Manager(target=o2), 671 Manager(target=o1) 672 ]) 673 session.commit() 674 eq_(session.query(Engineer).first().target, o1) 675 676 def test_joined_from_single(self): 677 678 class Company(Base, fixtures.ComparableEntity): 679 680 __tablename__ = 'companies' 681 id = Column('id', Integer, primary_key=True, 682 test_needs_autoincrement=True) 683 name = Column('name', String(50)) 684 employees = relationship('Person') 685 686 class Person(Base, fixtures.ComparableEntity): 687 688 __tablename__ = 'people' 689 id = Column(Integer, primary_key=True, 690 test_needs_autoincrement=True) 691 company_id = Column(Integer, ForeignKey('companies.id')) 692 name = Column(String(50)) 693 discriminator = Column('type', String(50)) 694 __mapper_args__ = {'polymorphic_on': discriminator} 695 696 class Manager(Person): 697 698 __mapper_args__ = {'polymorphic_identity': 'manager'} 699 golf_swing = Column(String(50)) 700 701 class Engineer(Person): 702 703 __tablename__ = 'engineers' 704 __mapper_args__ = {'polymorphic_identity': 'engineer'} 705 id = Column(Integer, ForeignKey('people.id'), 706 primary_key=True) 707 primary_language = Column(String(50)) 708 709 assert Person.__table__.c.golf_swing is not None 710 assert 'primary_language' not in Person.__table__.c 711 assert Engineer.__table__.c.primary_language is not None 712 assert Engineer.primary_language is not None 713 assert Manager.golf_swing is not None 714 assert not hasattr(Person, 'primary_language') 715 assert not hasattr(Person, 'golf_swing') 716 assert not hasattr(Engineer, 'golf_swing') 717 assert not hasattr(Manager, 'primary_language') 718 Base.metadata.create_all() 719 sess = create_session() 720 e1 = Engineer(name='dilbert', primary_language='java') 721 e2 = Engineer(name='wally', primary_language='c++') 722 m1 = Manager(name='dogbert', golf_swing='fore!') 723 c1 = Company(name='MegaCorp, Inc.', employees=[e1, e2, m1]) 724 e3 = Engineer(name='vlad', primary_language='cobol') 725 c2 = Company(name='Elbonia, Inc.', employees=[e3]) 726 sess.add(c1) 727 sess.add(c2) 728 sess.flush() 729 sess.expunge_all() 730 eq_(sess.query(Person).with_polymorphic(Engineer). 731 filter(Engineer.primary_language 732 == 'cobol').first(), Engineer(name='vlad')) 733 eq_(sess.query(Company).filter(Company.employees.of_type(Engineer). 734 any(Engineer.primary_language 735 == 'cobol')).first(), c2) 736 eq_(sess.query(Engineer).filter_by(primary_language='cobol' 737 ).one(), 738 Engineer(name='vlad', primary_language='cobol')) 739 740 def test_single_from_joined_colsonsub(self): 741 class Person(Base, fixtures.ComparableEntity): 742 743 __tablename__ = 'people' 744 id = Column(Integer, primary_key=True, 745 test_needs_autoincrement=True) 746 name = Column(String(50)) 747 discriminator = Column('type', String(50)) 748 __mapper_args__ = {'polymorphic_on': discriminator} 749 750 class Manager(Person): 751 __tablename__ = 'manager' 752 __mapper_args__ = {'polymorphic_identity': 'manager'} 753 id = Column(Integer, ForeignKey('people.id'), primary_key=True) 754 golf_swing = Column(String(50)) 755 756 class Boss(Manager): 757 boss_name = Column(String(50)) 758 759 is_( 760 Boss.__mapper__.column_attrs['boss_name'].columns[0], 761 Manager.__table__.c.boss_name 762 ) 763 764 def test_polymorphic_on_converted_from_inst(self): 765 class A(Base): 766 __tablename__ = 'A' 767 id = Column(Integer, primary_key=True) 768 discriminator = Column(String) 769 770 @declared_attr 771 def __mapper_args__(cls): 772 return { 773 'polymorphic_identity': cls.__name__, 774 'polymorphic_on': cls.discriminator 775 } 776 777 class B(A): 778 pass 779 is_(B.__mapper__.polymorphic_on, A.__table__.c.discriminator) 780 781 def test_add_deferred(self): 782 783 class Person(Base, fixtures.ComparableEntity): 784 785 __tablename__ = 'people' 786 id = Column('id', Integer, primary_key=True, 787 test_needs_autoincrement=True) 788 789 Person.name = deferred(Column(String(10))) 790 Base.metadata.create_all() 791 sess = create_session() 792 p = Person(name='ratbert') 793 sess.add(p) 794 sess.flush() 795 sess.expunge_all() 796 eq_(sess.query(Person).all(), [Person(name='ratbert')]) 797 sess.expunge_all() 798 person = sess.query(Person).filter(Person.name == 'ratbert' 799 ).one() 800 assert 'name' not in person.__dict__ 801 802 def test_single_fksonsub(self): 803 """test single inheritance with a foreign key-holding column on 804 a subclass. 805 806 """ 807 808 class Person(Base, fixtures.ComparableEntity): 809 810 __tablename__ = 'people' 811 id = Column(Integer, primary_key=True, 812 test_needs_autoincrement=True) 813 name = Column(String(50)) 814 discriminator = Column('type', String(50)) 815 __mapper_args__ = {'polymorphic_on': discriminator} 816 817 class Engineer(Person): 818 819 __mapper_args__ = {'polymorphic_identity': 'engineer'} 820 primary_language_id = Column(Integer, 821 ForeignKey('languages.id')) 822 primary_language = relationship('Language') 823 824 class Language(Base, fixtures.ComparableEntity): 825 826 __tablename__ = 'languages' 827 id = Column(Integer, primary_key=True, 828 test_needs_autoincrement=True) 829 name = Column(String(50)) 830 831 assert not hasattr(Person, 'primary_language_id') 832 Base.metadata.create_all() 833 sess = create_session() 834 java, cpp, cobol = Language(name='java'), Language(name='cpp'), \ 835 Language(name='cobol') 836 e1 = Engineer(name='dilbert', primary_language=java) 837 e2 = Engineer(name='wally', primary_language=cpp) 838 e3 = Engineer(name='vlad', primary_language=cobol) 839 sess.add_all([e1, e2, e3]) 840 sess.flush() 841 sess.expunge_all() 842 eq_(sess.query(Person).filter(Engineer.primary_language.has( 843 Language.name 844 == 'cobol')).first(), 845 Engineer(name='vlad', primary_language=Language(name='cobol'))) 846 eq_(sess.query(Engineer).filter(Engineer.primary_language.has( 847 Language.name 848 == 'cobol')).one(), 849 Engineer(name='vlad', primary_language=Language(name='cobol'))) 850 eq_(sess.query(Person).join(Engineer.primary_language).order_by( 851 Language.name).all(), 852 [Engineer(name='vlad', 853 primary_language=Language(name='cobol')), 854 Engineer(name='wally', primary_language=Language(name='cpp' 855 )), 856 Engineer(name='dilbert', primary_language=Language(name='java'))]) 857 858 def test_single_three_levels(self): 859 860 class Person(Base, fixtures.ComparableEntity): 861 862 __tablename__ = 'people' 863 id = Column(Integer, primary_key=True) 864 name = Column(String(50)) 865 discriminator = Column('type', String(50)) 866 __mapper_args__ = {'polymorphic_on': discriminator} 867 868 class Engineer(Person): 869 870 __mapper_args__ = {'polymorphic_identity': 'engineer'} 871 primary_language = Column(String(50)) 872 873 class JuniorEngineer(Engineer): 874 875 __mapper_args__ = \ 876 {'polymorphic_identity': 'junior_engineer'} 877 nerf_gun = Column(String(50)) 878 879 class Manager(Person): 880 881 __mapper_args__ = {'polymorphic_identity': 'manager'} 882 golf_swing = Column(String(50)) 883 884 assert JuniorEngineer.nerf_gun 885 assert JuniorEngineer.primary_language 886 assert JuniorEngineer.name 887 assert Manager.golf_swing 888 assert Engineer.primary_language 889 assert not hasattr(Engineer, 'golf_swing') 890 assert not hasattr(Engineer, 'nerf_gun') 891 assert not hasattr(Manager, 'nerf_gun') 892 assert not hasattr(Manager, 'primary_language') 893 894 def test_single_detects_conflict(self): 895 896 class Person(Base): 897 898 __tablename__ = 'people' 899 id = Column(Integer, primary_key=True) 900 name = Column(String(50)) 901 discriminator = Column('type', String(50)) 902 __mapper_args__ = {'polymorphic_on': discriminator} 903 904 class Engineer(Person): 905 906 __mapper_args__ = {'polymorphic_identity': 'engineer'} 907 primary_language = Column(String(50)) 908 909 # test sibling col conflict 910 911 def go(): 912 913 class Manager(Person): 914 915 __mapper_args__ = {'polymorphic_identity': 'manager'} 916 golf_swing = Column(String(50)) 917 primary_language = Column(String(50)) 918 919 assert_raises(sa.exc.ArgumentError, go) 920 921 # test parent col conflict 922 923 def go(): 924 925 class Salesman(Person): 926 927 __mapper_args__ = {'polymorphic_identity': 'manager'} 928 name = Column(String(50)) 929 930 assert_raises(sa.exc.ArgumentError, go) 931 932 def test_single_no_special_cols(self): 933 934 class Person(Base, fixtures.ComparableEntity): 935 936 __tablename__ = 'people' 937 id = Column('id', Integer, primary_key=True) 938 name = Column('name', String(50)) 939 discriminator = Column('type', String(50)) 940 __mapper_args__ = {'polymorphic_on': discriminator} 941 942 def go(): 943 944 class Engineer(Person): 945 946 __mapper_args__ = {'polymorphic_identity': 'engineer'} 947 primary_language = Column('primary_language', 948 String(50)) 949 foo_bar = Column(Integer, primary_key=True) 950 951 assert_raises_message(sa.exc.ArgumentError, 952 'place primary key', go) 953 954 def test_single_no_table_args(self): 955 956 class Person(Base, fixtures.ComparableEntity): 957 958 __tablename__ = 'people' 959 id = Column('id', Integer, primary_key=True) 960 name = Column('name', String(50)) 961 discriminator = Column('type', String(50)) 962 __mapper_args__ = {'polymorphic_on': discriminator} 963 964 def go(): 965 966 class Engineer(Person): 967 968 __mapper_args__ = {'polymorphic_identity': 'engineer'} 969 primary_language = Column('primary_language', 970 String(50)) 971 972 # this should be on the Person class, as this is single 973 # table inheritance, which is why we test that this 974 # throws an exception! 975 976 __table_args__ = {'mysql_engine': 'InnoDB'} 977 978 assert_raises_message(sa.exc.ArgumentError, 979 'place __table_args__', go) 980 981 @testing.emits_warning("This declarative") 982 def test_dupe_name_in_hierarchy(self): 983 class A(Base): 984 __tablename__ = "a" 985 id = Column(Integer, primary_key=True) 986 a_1 = A 987 988 class A(a_1): 989 __tablename__ = 'b' 990 id = Column(Integer(), ForeignKey(a_1.id), primary_key=True) 991 992 assert A.__mapper__.inherits is a_1.__mapper__ 993 994 995class OverlapColPrecedenceTest(DeclarativeTestBase): 996 997 """test #1892 cases when declarative does column precedence.""" 998 999 def _run_test(self, Engineer, e_id, p_id): 1000 p_table = Base.metadata.tables['person'] 1001 e_table = Base.metadata.tables['engineer'] 1002 assert Engineer.id.property.columns[0] is e_table.c[e_id] 1003 assert Engineer.id.property.columns[1] is p_table.c[p_id] 1004 1005 def test_basic(self): 1006 class Person(Base): 1007 __tablename__ = 'person' 1008 id = Column(Integer, primary_key=True) 1009 1010 class Engineer(Person): 1011 __tablename__ = 'engineer' 1012 id = Column(Integer, ForeignKey('person.id'), primary_key=True) 1013 1014 self._run_test(Engineer, "id", "id") 1015 1016 def test_alt_name_base(self): 1017 class Person(Base): 1018 __tablename__ = 'person' 1019 id = Column("pid", Integer, primary_key=True) 1020 1021 class Engineer(Person): 1022 __tablename__ = 'engineer' 1023 id = Column(Integer, ForeignKey('person.pid'), primary_key=True) 1024 1025 self._run_test(Engineer, "id", "pid") 1026 1027 def test_alt_name_sub(self): 1028 class Person(Base): 1029 __tablename__ = 'person' 1030 id = Column(Integer, primary_key=True) 1031 1032 class Engineer(Person): 1033 __tablename__ = 'engineer' 1034 id = Column("eid", Integer, ForeignKey('person.id'), 1035 primary_key=True) 1036 1037 self._run_test(Engineer, "eid", "id") 1038 1039 def test_alt_name_both(self): 1040 class Person(Base): 1041 __tablename__ = 'person' 1042 id = Column("pid", Integer, primary_key=True) 1043 1044 class Engineer(Person): 1045 __tablename__ = 'engineer' 1046 id = Column("eid", Integer, ForeignKey('person.pid'), 1047 primary_key=True) 1048 1049 self._run_test(Engineer, "eid", "pid") 1050 1051 1052class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase): 1053 1054 def _roundtrip(self, Employee, Manager, Engineer, Boss, 1055 polymorphic=True, explicit_type=False): 1056 Base.metadata.create_all() 1057 sess = create_session() 1058 e1 = Engineer(name='dilbert', primary_language='java') 1059 e2 = Engineer(name='wally', primary_language='c++') 1060 m1 = Manager(name='dogbert', golf_swing='fore!') 1061 e3 = Engineer(name='vlad', primary_language='cobol') 1062 b1 = Boss(name="pointy haired") 1063 1064 if polymorphic: 1065 for obj in [e1, e2, m1, e3, b1]: 1066 if explicit_type: 1067 eq_(obj.type, obj.__mapper__.polymorphic_identity) 1068 else: 1069 assert_raises_message( 1070 AttributeError, 1071 "does not implement attribute .?'type' " 1072 "at the instance level.", 1073 getattr, obj, "type" 1074 ) 1075 else: 1076 assert "type" not in Engineer.__dict__ 1077 assert "type" not in Manager.__dict__ 1078 assert "type" not in Boss.__dict__ 1079 1080 sess.add_all([e1, e2, m1, e3, b1]) 1081 sess.flush() 1082 sess.expunge_all() 1083 if polymorphic: 1084 eq_(sess.query(Employee).order_by(Employee.name).all(), 1085 [Engineer(name='dilbert'), Manager(name='dogbert'), 1086 Boss(name='pointy haired'), 1087 Engineer(name='vlad'), Engineer(name='wally')]) 1088 else: 1089 eq_(sess.query(Engineer).order_by(Engineer.name).all(), 1090 [Engineer(name='dilbert'), Engineer(name='vlad'), 1091 Engineer(name='wally')]) 1092 eq_(sess.query(Manager).all(), [Manager(name='dogbert')]) 1093 eq_(sess.query(Boss).all(), [Boss(name='pointy haired')]) 1094 1095 e1 = sess.query(Engineer).order_by(Engineer.name).first() 1096 sess.expire(e1) 1097 eq_(e1.name, 'dilbert') 1098 1099 def test_explicit(self): 1100 engineers = Table( 1101 'engineers', Base.metadata, 1102 Column('id', 1103 Integer, primary_key=True, test_needs_autoincrement=True), 1104 Column('name', String(50)), 1105 Column('primary_language', String(50))) 1106 managers = Table('managers', Base.metadata, 1107 Column('id', Integer, primary_key=True, 1108 test_needs_autoincrement=True), 1109 Column('name', String(50)), 1110 Column('golf_swing', String(50)) 1111 ) 1112 boss = Table('boss', Base.metadata, 1113 Column('id', Integer, primary_key=True, 1114 test_needs_autoincrement=True), 1115 Column('name', String(50)), 1116 Column('golf_swing', String(50)) 1117 ) 1118 punion = polymorphic_union({ 1119 'engineer': engineers, 1120 'manager': managers, 1121 'boss': boss}, 'type', 'punion') 1122 1123 class Employee(Base, fixtures.ComparableEntity): 1124 1125 __table__ = punion 1126 __mapper_args__ = {'polymorphic_on': punion.c.type} 1127 1128 class Engineer(Employee): 1129 1130 __table__ = engineers 1131 __mapper_args__ = {'polymorphic_identity': 'engineer', 1132 'concrete': True} 1133 1134 class Manager(Employee): 1135 1136 __table__ = managers 1137 __mapper_args__ = {'polymorphic_identity': 'manager', 1138 'concrete': True} 1139 1140 class Boss(Manager): 1141 __table__ = boss 1142 __mapper_args__ = {'polymorphic_identity': 'boss', 1143 'concrete': True} 1144 1145 self._roundtrip(Employee, Manager, Engineer, Boss) 1146 1147 def test_concrete_inline_non_polymorphic(self): 1148 """test the example from the declarative docs.""" 1149 1150 class Employee(Base, fixtures.ComparableEntity): 1151 1152 __tablename__ = 'people' 1153 id = Column(Integer, primary_key=True, 1154 test_needs_autoincrement=True) 1155 name = Column(String(50)) 1156 1157 class Engineer(Employee): 1158 1159 __tablename__ = 'engineers' 1160 __mapper_args__ = {'concrete': True} 1161 id = Column(Integer, primary_key=True, 1162 test_needs_autoincrement=True) 1163 primary_language = Column(String(50)) 1164 name = Column(String(50)) 1165 1166 class Manager(Employee): 1167 1168 __tablename__ = 'manager' 1169 __mapper_args__ = {'concrete': True} 1170 id = Column(Integer, primary_key=True, 1171 test_needs_autoincrement=True) 1172 golf_swing = Column(String(50)) 1173 name = Column(String(50)) 1174 1175 class Boss(Manager): 1176 __tablename__ = 'boss' 1177 __mapper_args__ = {'concrete': True} 1178 id = Column(Integer, primary_key=True, 1179 test_needs_autoincrement=True) 1180 golf_swing = Column(String(50)) 1181 name = Column(String(50)) 1182 1183 self._roundtrip(Employee, Manager, Engineer, Boss, polymorphic=False) 1184 1185 def test_abstract_concrete_extension(self): 1186 class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity): 1187 pass 1188 1189 class Manager(Employee): 1190 __tablename__ = 'manager' 1191 employee_id = Column(Integer, primary_key=True, 1192 test_needs_autoincrement=True) 1193 name = Column(String(50)) 1194 golf_swing = Column(String(40)) 1195 __mapper_args__ = { 1196 'polymorphic_identity': 'manager', 1197 'concrete': True} 1198 1199 class Boss(Manager): 1200 __tablename__ = 'boss' 1201 employee_id = Column(Integer, primary_key=True, 1202 test_needs_autoincrement=True) 1203 name = Column(String(50)) 1204 golf_swing = Column(String(40)) 1205 __mapper_args__ = { 1206 'polymorphic_identity': 'boss', 1207 'concrete': True} 1208 1209 class Engineer(Employee): 1210 __tablename__ = 'engineer' 1211 employee_id = Column(Integer, primary_key=True, 1212 test_needs_autoincrement=True) 1213 name = Column(String(50)) 1214 primary_language = Column(String(40)) 1215 __mapper_args__ = {'polymorphic_identity': 'engineer', 1216 'concrete': True} 1217 1218 self._roundtrip(Employee, Manager, Engineer, Boss) 1219 1220 def test_abstract_concrete_extension_descriptor_refresh(self): 1221 class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity): 1222 @declared_attr 1223 def name(cls): 1224 return Column(String(50)) 1225 1226 class Manager(Employee): 1227 __tablename__ = 'manager' 1228 employee_id = Column(Integer, primary_key=True, 1229 test_needs_autoincrement=True) 1230 paperwork = Column(String(10)) 1231 __mapper_args__ = { 1232 'polymorphic_identity': 'manager', 'concrete': True} 1233 1234 class Engineer(Employee): 1235 __tablename__ = 'engineer' 1236 employee_id = Column(Integer, primary_key=True, 1237 test_needs_autoincrement=True) 1238 1239 @property 1240 def paperwork(self): 1241 return "p" 1242 1243 __mapper_args__ = { 1244 'polymorphic_identity': 'engineer', 'concrete': True} 1245 1246 Base.metadata.create_all() 1247 sess = Session() 1248 sess.add(Engineer(name='d')) 1249 sess.commit() 1250 1251 # paperwork is excluded because there's a descritor; so it is 1252 # not in the Engineers mapped properties at all, though is inside the 1253 # class manager. Maybe it shouldn't be in the class manager either. 1254 assert 'paperwork' in Engineer.__mapper__.class_manager 1255 assert 'paperwork' not in Engineer.__mapper__.attrs.keys() 1256 1257 # type currently does get mapped, as a 1258 # ConcreteInheritedProperty, which means, "ignore this thing inherited 1259 # from the concrete base". if we didn't specify concrete=True, then 1260 # this one gets stuck in the error condition also. 1261 assert 'type' in Engineer.__mapper__.class_manager 1262 assert 'type' in Engineer.__mapper__.attrs.keys() 1263 1264 e1 = sess.query(Engineer).first() 1265 eq_(e1.name, 'd') 1266 sess.expire(e1) 1267 eq_(e1.name, 'd') 1268 1269 def test_concrete_extension(self): 1270 class Employee(ConcreteBase, Base, fixtures.ComparableEntity): 1271 __tablename__ = 'employee' 1272 employee_id = Column(Integer, primary_key=True, 1273 test_needs_autoincrement=True) 1274 name = Column(String(50)) 1275 __mapper_args__ = { 1276 'polymorphic_identity': 'employee', 1277 'concrete': True} 1278 1279 class Manager(Employee): 1280 __tablename__ = 'manager' 1281 employee_id = Column(Integer, primary_key=True, 1282 test_needs_autoincrement=True) 1283 name = Column(String(50)) 1284 golf_swing = Column(String(40)) 1285 __mapper_args__ = { 1286 'polymorphic_identity': 'manager', 1287 'concrete': True} 1288 1289 class Boss(Manager): 1290 __tablename__ = 'boss' 1291 employee_id = Column(Integer, primary_key=True, 1292 test_needs_autoincrement=True) 1293 name = Column(String(50)) 1294 golf_swing = Column(String(40)) 1295 __mapper_args__ = { 1296 'polymorphic_identity': 'boss', 1297 'concrete': True} 1298 1299 class Engineer(Employee): 1300 __tablename__ = 'engineer' 1301 employee_id = Column(Integer, primary_key=True, 1302 test_needs_autoincrement=True) 1303 name = Column(String(50)) 1304 primary_language = Column(String(40)) 1305 __mapper_args__ = {'polymorphic_identity': 'engineer', 1306 'concrete': True} 1307 self._roundtrip(Employee, Manager, Engineer, Boss) 1308 1309 def test_has_inherited_table_doesnt_consider_base(self): 1310 class A(Base): 1311 __tablename__ = 'a' 1312 id = Column(Integer, primary_key=True) 1313 1314 assert not has_inherited_table(A) 1315 1316 class B(A): 1317 __tablename__ = 'b' 1318 id = Column(Integer, ForeignKey('a.id'), primary_key=True) 1319 1320 assert has_inherited_table(B) 1321 1322 def test_has_inherited_table_in_mapper_args(self): 1323 class Test(Base): 1324 __tablename__ = 'test' 1325 id = Column(Integer, primary_key=True) 1326 type = Column(String(20)) 1327 1328 @declared_attr 1329 def __mapper_args__(cls): 1330 if not has_inherited_table(cls): 1331 ret = { 1332 'polymorphic_identity': 'default', 1333 'polymorphic_on': cls.type, 1334 } 1335 else: 1336 ret = {'polymorphic_identity': cls.__name__} 1337 return ret 1338 1339 class PolyTest(Test): 1340 __tablename__ = 'poly_test' 1341 id = Column(Integer, ForeignKey(Test.id), primary_key=True) 1342 1343 configure_mappers() 1344 1345 assert Test.__mapper__.polymorphic_on is Test.__table__.c.type 1346 assert PolyTest.__mapper__.polymorphic_on is Test.__table__.c.type 1347 1348 def test_ok_to_override_type_from_abstract(self): 1349 class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity): 1350 pass 1351 1352 class Manager(Employee): 1353 __tablename__ = 'manager' 1354 employee_id = Column(Integer, primary_key=True, 1355 test_needs_autoincrement=True) 1356 name = Column(String(50)) 1357 golf_swing = Column(String(40)) 1358 1359 @property 1360 def type(self): 1361 return "manager" 1362 1363 __mapper_args__ = { 1364 'polymorphic_identity': "manager", 1365 'concrete': True} 1366 1367 class Boss(Manager): 1368 __tablename__ = 'boss' 1369 employee_id = Column(Integer, primary_key=True, 1370 test_needs_autoincrement=True) 1371 name = Column(String(50)) 1372 golf_swing = Column(String(40)) 1373 1374 @property 1375 def type(self): 1376 return "boss" 1377 1378 __mapper_args__ = { 1379 'polymorphic_identity': "boss", 1380 'concrete': True} 1381 1382 class Engineer(Employee): 1383 __tablename__ = 'engineer' 1384 employee_id = Column(Integer, primary_key=True, 1385 test_needs_autoincrement=True) 1386 name = Column(String(50)) 1387 primary_language = Column(String(40)) 1388 1389 @property 1390 def type(self): 1391 return "engineer" 1392 __mapper_args__ = {'polymorphic_identity': "engineer", 1393 'concrete': True} 1394 self._roundtrip(Employee, Manager, Engineer, Boss, explicit_type=True) 1395 1396 1397class ConcreteExtensionConfigTest( 1398 _RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase): 1399 __dialect__ = 'default' 1400 1401 def test_classreg_setup(self): 1402 class A(Base, fixtures.ComparableEntity): 1403 __tablename__ = 'a' 1404 id = Column(Integer, 1405 primary_key=True, test_needs_autoincrement=True) 1406 data = Column(String(50)) 1407 collection = relationship("BC", primaryjoin="BC.a_id == A.id", 1408 collection_class=set) 1409 1410 class BC(AbstractConcreteBase, Base, fixtures.ComparableEntity): 1411 pass 1412 1413 class B(BC): 1414 __tablename__ = 'b' 1415 id = Column(Integer, 1416 primary_key=True, test_needs_autoincrement=True) 1417 1418 a_id = Column(Integer, ForeignKey('a.id')) 1419 data = Column(String(50)) 1420 b_data = Column(String(50)) 1421 __mapper_args__ = { 1422 "polymorphic_identity": "b", 1423 "concrete": True 1424 } 1425 1426 class C(BC): 1427 __tablename__ = 'c' 1428 id = Column(Integer, 1429 primary_key=True, test_needs_autoincrement=True) 1430 a_id = Column(Integer, ForeignKey('a.id')) 1431 data = Column(String(50)) 1432 c_data = Column(String(50)) 1433 __mapper_args__ = { 1434 "polymorphic_identity": "c", 1435 "concrete": True 1436 } 1437 1438 Base.metadata.create_all() 1439 sess = Session() 1440 sess.add_all([ 1441 A(data='a1', collection=set([ 1442 B(data='a1b1', b_data='a1b1'), 1443 C(data='a1b2', c_data='a1c1'), 1444 B(data='a1b2', b_data='a1b2'), 1445 C(data='a1c2', c_data='a1c2'), 1446 ])), 1447 A(data='a2', collection=set([ 1448 B(data='a2b1', b_data='a2b1'), 1449 C(data='a2c1', c_data='a2c1'), 1450 B(data='a2b2', b_data='a2b2'), 1451 C(data='a2c2', c_data='a2c2'), 1452 ])) 1453 ]) 1454 sess.commit() 1455 sess.expunge_all() 1456 1457 eq_( 1458 sess.query(A).filter_by(data='a2').all(), 1459 [ 1460 A(data='a2', collection=set([ 1461 B(data='a2b1', b_data='a2b1'), 1462 B(data='a2b2', b_data='a2b2'), 1463 C(data='a2c1', c_data='a2c1'), 1464 C(data='a2c2', c_data='a2c2'), 1465 ])) 1466 ] 1467 ) 1468 1469 self.assert_compile( 1470 sess.query(A).join(A.collection), 1471 "SELECT a.id AS a_id, a.data AS a_data FROM a JOIN " 1472 "(SELECT c.id AS id, c.a_id AS a_id, c.data AS data, " 1473 "c.c_data AS c_data, CAST(NULL AS VARCHAR(50)) AS b_data, " 1474 "'c' AS type FROM c UNION ALL SELECT b.id AS id, b.a_id AS a_id, " 1475 "b.data AS data, CAST(NULL AS VARCHAR(50)) AS c_data, " 1476 "b.b_data AS b_data, 'b' AS type FROM b) AS pjoin " 1477 "ON pjoin.a_id = a.id" 1478 ) 1479 1480 def test_prop_on_base(self): 1481 """test [ticket:2670] """ 1482 1483 counter = mock.Mock() 1484 1485 class Something(Base): 1486 __tablename__ = 'something' 1487 id = Column(Integer, primary_key=True) 1488 1489 class AbstractConcreteAbstraction(AbstractConcreteBase, Base): 1490 id = Column(Integer, primary_key=True) 1491 x = Column(Integer) 1492 y = Column(Integer) 1493 1494 @declared_attr 1495 def something_id(cls): 1496 return Column(ForeignKey(Something.id)) 1497 1498 @declared_attr 1499 def something(cls): 1500 counter(cls, "something") 1501 return relationship("Something") 1502 1503 @declared_attr 1504 def something_else(cls): 1505 counter(cls, "something_else") 1506 return relationship("Something") 1507 1508 class ConcreteConcreteAbstraction(AbstractConcreteAbstraction): 1509 __tablename__ = 'cca' 1510 __mapper_args__ = { 1511 'polymorphic_identity': 'ccb', 1512 'concrete': True} 1513 1514 # concrete is mapped, the abstract base is not (yet) 1515 assert ConcreteConcreteAbstraction.__mapper__ 1516 assert not hasattr(AbstractConcreteAbstraction, '__mapper__') 1517 1518 session = Session() 1519 self.assert_compile( 1520 session.query(ConcreteConcreteAbstraction).filter( 1521 ConcreteConcreteAbstraction.something.has(id=1)), 1522 "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, " 1523 "cca.something_id AS cca_something_id FROM cca WHERE EXISTS " 1524 "(SELECT 1 FROM something WHERE something.id = cca.something_id " 1525 "AND something.id = :id_1)" 1526 ) 1527 1528 # now it is 1529 assert AbstractConcreteAbstraction.__mapper__ 1530 1531 self.assert_compile( 1532 session.query(ConcreteConcreteAbstraction).filter( 1533 ConcreteConcreteAbstraction.something_else.has(id=1)), 1534 "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, " 1535 "cca.something_id AS cca_something_id FROM cca WHERE EXISTS " 1536 "(SELECT 1 FROM something WHERE something.id = cca.something_id " 1537 "AND something.id = :id_1)" 1538 ) 1539 1540 self.assert_compile( 1541 session.query(AbstractConcreteAbstraction).filter( 1542 AbstractConcreteAbstraction.something.has(id=1)), 1543 "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, " 1544 "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, " 1545 "pjoin.type AS pjoin_type FROM " 1546 "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, " 1547 "cca.something_id AS something_id, 'ccb' AS type FROM cca) " 1548 "AS pjoin WHERE EXISTS (SELECT 1 FROM something " 1549 "WHERE something.id = pjoin.something_id AND something.id = :id_1)" 1550 ) 1551 1552 self.assert_compile( 1553 session.query(AbstractConcreteAbstraction).filter( 1554 AbstractConcreteAbstraction.something_else.has(id=1)), 1555 "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, " 1556 "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, " 1557 "pjoin.type AS pjoin_type FROM " 1558 "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, " 1559 "cca.something_id AS something_id, 'ccb' AS type FROM cca) " 1560 "AS pjoin WHERE EXISTS (SELECT 1 FROM something " 1561 "WHERE something.id = pjoin.something_id AND something.id = :id_1)" 1562 ) 1563 1564 def test_abstract_in_hierarchy(self): 1565 class Document(Base, AbstractConcreteBase): 1566 doctype = Column(String) 1567 1568 class ContactDocument(Document): 1569 __abstract__ = True 1570 1571 send_method = Column(String) 1572 1573 class ActualDocument(ContactDocument): 1574 __tablename__ = 'actual_documents' 1575 __mapper_args__ = { 1576 'concrete': True, 1577 'polymorphic_identity': 'actual'} 1578 1579 id = Column(Integer, primary_key=True) 1580 1581 configure_mappers() 1582 session = Session() 1583 self.assert_compile( 1584 session.query(Document), 1585 "SELECT pjoin.doctype AS pjoin_doctype, " 1586 "pjoin.send_method AS pjoin_send_method, " 1587 "pjoin.id AS pjoin_id, pjoin.type AS pjoin_type " 1588 "FROM (SELECT actual_documents.doctype AS doctype, " 1589 "actual_documents.send_method AS send_method, " 1590 "actual_documents.id AS id, 'actual' AS type " 1591 "FROM actual_documents) AS pjoin" 1592 ) 1593 1594 def test_column_attr_names(self): 1595 """test #3480""" 1596 1597 class Document(Base, AbstractConcreteBase): 1598 documentType = Column('documenttype', String) 1599 1600 class Offer(Document): 1601 __tablename__ = 'offers' 1602 1603 id = Column(Integer, primary_key=True) 1604 __mapper_args__ = { 1605 'polymorphic_identity': 'offer' 1606 } 1607 1608 configure_mappers() 1609 session = Session() 1610 self.assert_compile( 1611 session.query(Document), 1612 "SELECT pjoin.documenttype AS pjoin_documenttype, " 1613 "pjoin.id AS pjoin_id, pjoin.type AS pjoin_type FROM " 1614 "(SELECT offers.documenttype AS documenttype, offers.id AS id, " 1615 "'offer' AS type FROM offers) AS pjoin" 1616 ) 1617 1618 self.assert_compile( 1619 session.query(Document.documentType), 1620 "SELECT pjoin.documenttype AS pjoin_documenttype FROM " 1621 "(SELECT offers.documenttype AS documenttype, offers.id AS id, " 1622 "'offer' AS type FROM offers) AS pjoin" 1623 ) 1624