1from sqlalchemy import desc 2from sqlalchemy import exc as sa_exc 3from sqlalchemy import func 4from sqlalchemy import select 5from sqlalchemy import testing 6from sqlalchemy import true 7from sqlalchemy.orm import aliased 8from sqlalchemy.orm import defaultload 9from sqlalchemy.orm import join 10from sqlalchemy.orm import joinedload 11from sqlalchemy.orm import selectinload 12from sqlalchemy.orm import subqueryload 13from sqlalchemy.orm import with_polymorphic 14from sqlalchemy.testing import assert_raises 15from sqlalchemy.testing import eq_ 16from sqlalchemy.testing import fixtures 17from sqlalchemy.testing.assertsql import CompiledSQL 18from sqlalchemy.testing.fixtures import fixture_session 19from ._poly_fixtures import _Polymorphic 20from ._poly_fixtures import _PolymorphicAliasedJoins 21from ._poly_fixtures import _PolymorphicJoins 22from ._poly_fixtures import _PolymorphicPolymorphic 23from ._poly_fixtures import _PolymorphicUnions 24from ._poly_fixtures import Boss 25from ._poly_fixtures import Company 26from ._poly_fixtures import Engineer 27from ._poly_fixtures import Machine 28from ._poly_fixtures import Manager 29from ._poly_fixtures import Paperwork 30from ._poly_fixtures import Person 31 32 33class _PolymorphicTestBase(fixtures.NoCache): 34 __backend__ = True 35 __dialect__ = "default_enhanced" 36 37 @classmethod 38 def setup_mappers(cls): 39 super(_PolymorphicTestBase, cls).setup_mappers() 40 global people, engineers, managers, boss 41 global companies, paperwork, machines 42 people, engineers, managers, boss, companies, paperwork, machines = ( 43 cls.tables.people, 44 cls.tables.engineers, 45 cls.tables.managers, 46 cls.tables.boss, 47 cls.tables.companies, 48 cls.tables.paperwork, 49 cls.tables.machines, 50 ) 51 52 @classmethod 53 def insert_data(cls, connection): 54 super(_PolymorphicTestBase, cls).insert_data(connection) 55 56 global all_employees, c1_employees, c2_employees 57 global c1, c2, e1, e2, e3, b1, m1 58 c1, c2, all_employees, c1_employees, c2_employees = ( 59 cls.c1, 60 cls.c2, 61 cls.all_employees, 62 cls.c1_employees, 63 cls.c2_employees, 64 ) 65 e1, e2, e3, b1, m1 = cls.e1, cls.e2, cls.e3, cls.b1, cls.m1 66 67 def test_loads_at_once(self): 68 """ 69 Test that all objects load from the full query, when 70 with_polymorphic is used. 71 """ 72 73 sess = fixture_session() 74 75 def go(): 76 eq_( 77 sess.query(Person).order_by(Person.person_id).all(), 78 all_employees, 79 ) 80 81 count = {"": 14, "Polymorphic": 9}.get(self.select_type, 10) 82 self.assert_sql_count(testing.db, go, count) 83 84 def test_primary_eager_aliasing_joinedload(self): 85 # For both joinedload() and subqueryload(), if the original q is 86 # not loading the subclass table, the joinedload doesn't happen. 87 88 sess = fixture_session() 89 90 def go(): 91 eq_( 92 sess.query(Person) 93 .order_by(Person.person_id) 94 .options(joinedload(Engineer.machines))[1:3], 95 all_employees[1:3], 96 ) 97 98 count = {"": 6, "Polymorphic": 3}.get(self.select_type, 4) 99 self.assert_sql_count(testing.db, go, count) 100 101 def test_primary_eager_aliasing_subqueryload(self): 102 # test that subqueryload does not occur because the parent 103 # row cannot support it 104 sess = fixture_session() 105 106 def go(): 107 eq_( 108 sess.query(Person) 109 .order_by(Person.person_id) 110 .options(subqueryload(Engineer.machines)) 111 .all(), 112 all_employees, 113 ) 114 115 count = {"": 14, "Polymorphic": 7}.get(self.select_type, 8) 116 self.assert_sql_count(testing.db, go, count) 117 118 def test_primary_eager_aliasing_selectinload(self): 119 # test that selectinload does not occur because the parent 120 # row cannot support it 121 sess = fixture_session() 122 123 def go(): 124 eq_( 125 sess.query(Person) 126 .order_by(Person.person_id) 127 .options(selectinload(Engineer.machines)) 128 .all(), 129 all_employees, 130 ) 131 132 count = {"": 14, "Polymorphic": 7}.get(self.select_type, 8) 133 self.assert_sql_count(testing.db, go, count) 134 135 def test_primary_eager_aliasing_three_reset_selectable(self): 136 """test now related to #7262 137 138 See test_primary_eager_aliasing_three_dont_reset_selectable for the 139 non-reset selectable version. 140 141 """ 142 # assert the JOINs don't over JOIN 143 144 sess = fixture_session() 145 146 # note selectable=None 147 wp = with_polymorphic(Person, "*", None) 148 149 def go(): 150 eq_( 151 sess.query(wp) 152 .order_by(wp.person_id) 153 .options(joinedload(wp.Engineer.machines))[1:3], 154 all_employees[1:3], 155 ) 156 157 self.assert_sql_count(testing.db, go, 3) 158 159 eq_( 160 sess.scalar( 161 select(func.count("*")).select_from( 162 sess.query(wp) 163 .options(joinedload(wp.Engineer.machines)) 164 .order_by(wp.person_id) 165 .limit(2) 166 .offset(1) 167 .subquery() 168 ) 169 ), 170 2, 171 ) 172 173 def test_get_one(self): 174 """ 175 For all mappers, ensure the primary key has been calculated as 176 just the "person_id" column. 177 """ 178 sess = fixture_session() 179 eq_( 180 sess.get(Person, e1.person_id), 181 Engineer(name="dilbert", primary_language="java"), 182 ) 183 184 def test_get_two(self): 185 sess = fixture_session() 186 eq_( 187 sess.get(Engineer, e1.person_id), 188 Engineer(name="dilbert", primary_language="java"), 189 ) 190 191 def test_get_three(self): 192 sess = fixture_session() 193 eq_( 194 sess.get(Manager, b1.person_id), 195 Boss(name="pointy haired boss", golf_swing="fore"), 196 ) 197 198 def test_multi_join(self): 199 sess = fixture_session() 200 e = aliased(Person) 201 c = aliased(Company) 202 q = ( 203 sess.query(Company, Person, c, e) 204 .join(Person, Company.employees) 205 .join(e, c.employees) 206 .filter(Person.person_id != e.person_id) 207 .filter(Person.name == "dilbert") 208 .filter(e.name == "wally") 209 ) 210 eq_(q.count(), 1) 211 eq_( 212 q.all(), 213 [ 214 ( 215 Company(company_id=1, name="MegaCorp, Inc."), 216 Engineer( 217 status="regular engineer", 218 engineer_name="dilbert", 219 name="dilbert", 220 company_id=1, 221 primary_language="java", 222 person_id=1, 223 type="engineer", 224 ), 225 Company(company_id=1, name="MegaCorp, Inc."), 226 Engineer( 227 status="regular engineer", 228 engineer_name="wally", 229 name="wally", 230 company_id=1, 231 primary_language="c++", 232 person_id=2, 233 type="engineer", 234 ), 235 ) 236 ], 237 ) 238 239 def test_multi_join_future(self): 240 sess = fixture_session(future=True) 241 e = aliased(Person) 242 c = aliased(Company) 243 244 q = ( 245 select(Company, Person, c, e) 246 .join(Person, Company.employees) 247 .join(e, c.employees) 248 .filter(Person.person_id != e.person_id) 249 .filter(Person.name == "dilbert") 250 .filter(e.name == "wally") 251 ) 252 253 eq_( 254 sess.execute( 255 select(func.count()).select_from(q.subquery()) 256 ).scalar(), 257 1, 258 ) 259 260 eq_( 261 sess.execute(q).all(), 262 [ 263 ( 264 Company(company_id=1, name="MegaCorp, Inc."), 265 Engineer( 266 status="regular engineer", 267 engineer_name="dilbert", 268 name="dilbert", 269 company_id=1, 270 primary_language="java", 271 person_id=1, 272 type="engineer", 273 ), 274 Company(company_id=1, name="MegaCorp, Inc."), 275 Engineer( 276 status="regular engineer", 277 engineer_name="wally", 278 name="wally", 279 company_id=1, 280 primary_language="c++", 281 person_id=2, 282 type="engineer", 283 ), 284 ) 285 ], 286 ) 287 288 def test_filter_on_subclass_one(self): 289 sess = fixture_session() 290 eq_(sess.query(Engineer).all()[0], Engineer(name="dilbert")) 291 292 def test_filter_on_subclass_one_future(self): 293 sess = fixture_session(future=True) 294 eq_( 295 sess.execute(select(Engineer)).scalar(), 296 Engineer(name="dilbert"), 297 ) 298 299 def test_filter_on_subclass_two(self): 300 sess = fixture_session() 301 eq_(sess.query(Engineer).first(), Engineer(name="dilbert")) 302 303 def test_filter_on_subclass_three(self): 304 sess = fixture_session() 305 eq_( 306 sess.query(Engineer) 307 .filter(Engineer.person_id == e1.person_id) 308 .first(), 309 Engineer(name="dilbert"), 310 ) 311 312 def test_filter_on_subclass_four(self): 313 sess = fixture_session() 314 eq_( 315 sess.query(Manager) 316 .filter(Manager.person_id == m1.person_id) 317 .one(), 318 Manager(name="dogbert"), 319 ) 320 321 def test_filter_on_subclass_five(self): 322 sess = fixture_session() 323 eq_( 324 sess.query(Manager) 325 .filter(Manager.person_id == b1.person_id) 326 .one(), 327 Boss(name="pointy haired boss"), 328 ) 329 330 def test_filter_on_subclass_six(self): 331 sess = fixture_session() 332 eq_( 333 sess.query(Boss).filter(Boss.person_id == b1.person_id).one(), 334 Boss(name="pointy haired boss"), 335 ) 336 337 def test_join_from_polymorphic_nonaliased_one(self): 338 sess = fixture_session() 339 eq_( 340 sess.query(Person) 341 .join(Person.paperwork) 342 .filter(Paperwork.description.like("%review%")) 343 .all(), 344 [b1, m1], 345 ) 346 347 def test_join_from_polymorphic_nonaliased_one_future(self): 348 sess = fixture_session(future=True) 349 eq_( 350 sess.execute( 351 select(Person) 352 .join(Person.paperwork) 353 .filter(Paperwork.description.like("%review%")) 354 ) 355 .unique() 356 .scalars() 357 .all(), 358 [b1, m1], 359 ) 360 361 def test_join_from_polymorphic_nonaliased_two(self): 362 sess = fixture_session() 363 eq_( 364 sess.query(Person) 365 .order_by(Person.person_id) 366 .join(Person.paperwork) 367 .filter(Paperwork.description.like("%#2%")) 368 .all(), 369 [e1, m1], 370 ) 371 372 def test_join_from_polymorphic_nonaliased_three(self): 373 sess = fixture_session() 374 eq_( 375 sess.query(Engineer) 376 .order_by(Person.person_id) 377 .join(Person.paperwork) 378 .filter(Paperwork.description.like("%#2%")) 379 .all(), 380 [e1], 381 ) 382 383 def test_join_from_polymorphic_nonaliased_four(self): 384 sess = fixture_session() 385 eq_( 386 sess.query(Person) 387 .order_by(Person.person_id) 388 .join(Person.paperwork) 389 .filter(Person.name.like("%dog%")) 390 .filter(Paperwork.description.like("%#2%")) 391 .all(), 392 [m1], 393 ) 394 395 def test_join_from_polymorphic_aliased_one_future(self): 396 sess = fixture_session(future=True) 397 398 pa = aliased(Paperwork) 399 eq_( 400 sess.execute( 401 select(Person) 402 .order_by(Person.person_id) 403 .join(Person.paperwork.of_type(pa)) 404 .filter(pa.description.like("%review%")) 405 ) 406 .unique() 407 .scalars() 408 .all(), 409 [b1, m1], 410 ) 411 412 def test_join_from_polymorphic_explicit_aliased_one(self): 413 sess = fixture_session() 414 pa = aliased(Paperwork) 415 eq_( 416 sess.query(Person) 417 .order_by(Person.person_id) 418 .join(pa, Person.paperwork) 419 .filter(pa.description.like("%review%")) 420 .all(), 421 [b1, m1], 422 ) 423 424 def test_join_from_polymorphic_explicit_aliased_two(self): 425 sess = fixture_session() 426 pa = aliased(Paperwork) 427 eq_( 428 sess.query(Person) 429 .order_by(Person.person_id) 430 .join(pa, Person.paperwork) 431 .filter(pa.description.like("%#2%")) 432 .all(), 433 [e1, m1], 434 ) 435 436 def test_join_from_polymorphic_explicit_aliased_three(self): 437 sess = fixture_session() 438 pa = aliased(Paperwork) 439 eq_( 440 sess.query(Engineer) 441 .order_by(Person.person_id) 442 .join(pa, Person.paperwork) 443 .filter(pa.description.like("%#2%")) 444 .all(), 445 [e1], 446 ) 447 448 def test_join_from_polymorphic_aliased_four(self): 449 sess = fixture_session() 450 pa = aliased(Paperwork) 451 eq_( 452 sess.query(Person) 453 .order_by(Person.person_id) 454 .join(pa, Person.paperwork) 455 .filter(Person.name.like("%dog%")) 456 .filter(pa.description.like("%#2%")) 457 .all(), 458 [m1], 459 ) 460 461 def test_join_from_with_polymorphic_nonaliased_one_future(self): 462 sess = fixture_session(future=True) 463 464 pm = with_polymorphic(Person, [Manager]) 465 eq_( 466 sess.execute( 467 select(pm) 468 .order_by(pm.person_id) 469 .join(pm.paperwork) 470 .filter(Paperwork.description.like("%review%")) 471 ) 472 .unique() 473 .scalars() 474 .all(), 475 [b1, m1], 476 ) 477 478 def test_join_from_with_polymorphic_nonaliased_two_future(self): 479 sess = fixture_session() 480 481 wp = with_polymorphic(Person, [Manager, Engineer]) 482 eq_( 483 sess.query(wp) 484 .order_by(wp.person_id) 485 .join(wp.paperwork) 486 .filter(Paperwork.description.like("%#2%")) 487 .all(), 488 [e1, m1], 489 ) 490 491 def test_join_from_with_polymorphic_nonaliased_three_future(self): 492 sess = fixture_session() 493 494 wp = with_polymorphic(Person, [Manager, Engineer]) 495 eq_( 496 sess.query(wp) 497 .order_by(wp.person_id) 498 .join(wp.paperwork) 499 .filter(wp.name.like("%dog%")) 500 .filter(Paperwork.description.like("%#2%")) 501 .all(), 502 [m1], 503 ) 504 505 def test_join_from_with_polymorphic_explicit_aliased_one_future(self): 506 sess = fixture_session() 507 pa = aliased(Paperwork) 508 wp = with_polymorphic(Person, [Manager]) 509 510 eq_( 511 sess.query(wp) 512 .join(pa, wp.paperwork) 513 .filter(pa.description.like("%review%")) 514 .all(), 515 [b1, m1], 516 ) 517 518 def test_join_from_with_polymorphic_explicit_aliased_two_future(self): 519 sess = fixture_session() 520 pa = aliased(Paperwork) 521 522 wp = with_polymorphic(Person, [Manager, Engineer]) 523 eq_( 524 sess.query(wp) 525 .order_by(wp.person_id) 526 .join(pa, wp.paperwork) 527 .filter(pa.description.like("%#2%")) 528 .all(), 529 [e1, m1], 530 ) 531 532 def test_join_from_with_polymorphic_ot_explicit_aliased_two_future(self): 533 sess = fixture_session() 534 pa = aliased(Paperwork) 535 536 wp = with_polymorphic(Person, [Manager, Engineer]) 537 eq_( 538 sess.query(wp) 539 .order_by(wp.person_id) 540 .join(wp.paperwork.of_type(pa)) 541 .filter(pa.description.like("%#2%")) 542 .all(), 543 [e1, m1], 544 ) 545 546 def test_join_from_with_polymorphic_aliased_three_future(self): 547 sess = fixture_session() 548 pa = aliased(Paperwork) 549 wp = with_polymorphic(Person, [Manager, Engineer]) 550 551 eq_( 552 sess.query(wp) 553 .order_by(wp.person_id) 554 .join(pa, wp.paperwork) 555 .filter(wp.name.like("%dog%")) 556 .filter(pa.description.like("%#2%")) 557 .all(), 558 [m1], 559 ) 560 561 def test_join_from_with_polymorphic_ot_aliased_three_future(self): 562 sess = fixture_session() 563 pa = aliased(Paperwork) 564 wp = with_polymorphic(Person, [Manager, Engineer]) 565 566 eq_( 567 sess.query(wp) 568 .order_by(wp.person_id) 569 .join(wp.paperwork.of_type(pa)) 570 .filter(wp.name.like("%dog%")) 571 .filter(pa.description.like("%#2%")) 572 .all(), 573 [m1], 574 ) 575 576 def test_join_to_polymorphic_nonaliased(self): 577 sess = fixture_session() 578 eq_( 579 sess.query(Company) 580 .join(Company.employees) 581 .filter(Person.name == "vlad") 582 .one(), 583 c2, 584 ) 585 586 def test_join_to_polymorphic_explicit_aliased(self): 587 sess = fixture_session() 588 ea = aliased(Person) 589 eq_( 590 sess.query(Company) 591 .join(ea, Company.employees) 592 .filter(ea.name == "vlad") 593 .one(), 594 c2, 595 ) 596 597 def test_polymorphic_any_one(self): 598 sess = fixture_session() 599 600 any_ = Company.employees.any(Person.name == "vlad") 601 eq_(sess.query(Company).filter(any_).all(), [c2]) 602 603 def test_polymorphic_any_explicit_alias_two(self): 604 sess = fixture_session() 605 # test that the aliasing on "Person" does not bleed into the 606 # EXISTS clause generated by any() 607 any_ = Company.employees.any(Person.name == "wally") 608 ea = aliased(Person) 609 eq_( 610 sess.query(Company) 611 .join(ea, Company.employees) 612 .filter(ea.name == "dilbert") 613 .filter(any_) 614 .all(), 615 [c1], 616 ) 617 618 def test_polymorphic_any_three(self): 619 sess = fixture_session() 620 any_ = Company.employees.any(Person.name == "vlad") 621 ea = aliased(Person) 622 eq_( 623 sess.query(Company) 624 .join(ea, Company.employees) 625 .filter(ea.name == "dilbert") 626 .filter(any_) 627 .all(), 628 [], 629 ) 630 631 def test_polymorphic_any_eight(self): 632 sess = fixture_session() 633 any_ = Engineer.machines.any(Machine.name == "Commodore 64") 634 eq_( 635 sess.query(Person).order_by(Person.person_id).filter(any_).all(), 636 [e2, e3], 637 ) 638 639 def test_polymorphic_any_nine(self): 640 sess = fixture_session() 641 any_ = Person.paperwork.any(Paperwork.description == "review #2") 642 eq_( 643 sess.query(Person).order_by(Person.person_id).filter(any_).all(), 644 [m1], 645 ) 646 647 def test_join_from_columns_or_subclass_one(self): 648 sess = fixture_session() 649 650 expected = [("dogbert",), ("pointy haired boss",)] 651 eq_(sess.query(Manager.name).order_by(Manager.name).all(), expected) 652 653 def test_join_from_columns_or_subclass_two(self): 654 sess = fixture_session() 655 expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)] 656 eq_( 657 sess.query(Manager.name) 658 .join(Paperwork, Manager.paperwork) 659 .order_by(Manager.name) 660 .all(), 661 expected, 662 ) 663 664 def test_join_from_columns_or_subclass_three(self): 665 sess = fixture_session() 666 expected = [ 667 ("dilbert",), 668 ("dilbert",), 669 ("dogbert",), 670 ("dogbert",), 671 ("pointy haired boss",), 672 ("vlad",), 673 ("wally",), 674 ("wally",), 675 ] 676 eq_( 677 sess.query(Person.name) 678 .join(Paperwork, Person.paperwork) 679 .order_by(Person.name) 680 .all(), 681 expected, 682 ) 683 684 def test_join_from_columns_or_subclass_four(self): 685 sess = fixture_session() 686 # Load Person.name, joining from Person -> paperwork, get all 687 # the people. 688 expected = [ 689 ("dilbert",), 690 ("dilbert",), 691 ("dogbert",), 692 ("dogbert",), 693 ("pointy haired boss",), 694 ("vlad",), 695 ("wally",), 696 ("wally",), 697 ] 698 eq_( 699 sess.query(Person.name) 700 .join(paperwork, Person.person_id == paperwork.c.person_id) 701 .order_by(Person.name) 702 .all(), 703 expected, 704 ) 705 706 def test_join_from_columns_or_subclass_five(self): 707 sess = fixture_session() 708 # same, on manager. get only managers. 709 expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)] 710 eq_( 711 sess.query(Manager.name) 712 .join(paperwork, Manager.person_id == paperwork.c.person_id) 713 .order_by(Person.name) 714 .all(), 715 expected, 716 ) 717 718 def test_join_from_columns_or_subclass_six(self): 719 sess = fixture_session() 720 if self.select_type == "": 721 # this now raises, due to [ticket:1892]. Manager.person_id 722 # is now the "person_id" column on Manager. SQL is incorrect. 723 assert_raises( 724 sa_exc.DBAPIError, 725 sess.query(Person.name) 726 .join(paperwork, Manager.person_id == paperwork.c.person_id) 727 .order_by(Person.name) 728 .all, 729 ) 730 elif self.select_type == "Unions": 731 # with the union, not something anyone would really be using 732 # here, it joins to the full result set. This is 0.6's 733 # behavior and is more or less wrong. 734 expected = [ 735 ("dilbert",), 736 ("dilbert",), 737 ("dogbert",), 738 ("dogbert",), 739 ("pointy haired boss",), 740 ("vlad",), 741 ("wally",), 742 ("wally",), 743 ] 744 eq_( 745 sess.query(Person.name) 746 .join(paperwork, Manager.person_id == paperwork.c.person_id) 747 .order_by(Person.name) 748 .all(), 749 expected, 750 ) 751 else: 752 # when a join is present and managers.person_id is available, 753 # you get the managers. 754 expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)] 755 eq_( 756 sess.query(Person.name) 757 .join(paperwork, Manager.person_id == paperwork.c.person_id) 758 .order_by(Person.name) 759 .all(), 760 expected, 761 ) 762 763 def test_join_from_columns_or_subclass_seven(self): 764 sess = fixture_session() 765 eq_( 766 sess.query(Manager) 767 .join(Paperwork, Manager.paperwork) 768 .order_by(Manager.name) 769 .all(), 770 [m1, b1], 771 ) 772 773 def test_join_from_columns_or_subclass_eight(self): 774 sess = fixture_session() 775 expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)] 776 eq_( 777 sess.query(Manager.name) 778 .join(paperwork, Manager.person_id == paperwork.c.person_id) 779 .order_by(Manager.name) 780 .all(), 781 expected, 782 ) 783 784 def test_join_from_columns_or_subclass_nine(self): 785 sess = fixture_session() 786 eq_( 787 sess.query(Manager.person_id) 788 .join(paperwork, Manager.person_id == paperwork.c.person_id) 789 .order_by(Manager.name) 790 .all(), 791 [(4,), (4,), (3,)], 792 ) 793 794 def test_join_from_columns_or_subclass_ten(self): 795 sess = fixture_session() 796 expected = [ 797 ("pointy haired boss", "review #1"), 798 ("dogbert", "review #2"), 799 ("dogbert", "review #3"), 800 ] 801 eq_( 802 sess.query(Manager.name, Paperwork.description) 803 .join(Paperwork, Manager.person_id == Paperwork.person_id) 804 .order_by(Paperwork.paperwork_id) 805 .all(), 806 expected, 807 ) 808 809 def test_join_from_columns_or_subclass_eleven(self): 810 sess = fixture_session() 811 expected = [("pointy haired boss",), ("dogbert",), ("dogbert",)] 812 malias = aliased(Manager) 813 eq_( 814 sess.query(malias.name) 815 .join(paperwork, malias.person_id == paperwork.c.person_id) 816 .all(), 817 expected, 818 ) 819 820 def test_subclass_option_pathing(self): 821 sess = fixture_session() 822 dilbert = ( 823 sess.query(Person) 824 .options(defaultload(Engineer.machines).defer(Machine.name)) 825 .filter(Person.name == "dilbert") 826 .first() 827 ) 828 m = dilbert.machines[0] 829 assert "name" not in m.__dict__ 830 eq_(m.name, "IBM ThinkPad") 831 832 def test_expire(self): 833 """ 834 Test that individual column refresh doesn't get tripped up by 835 the select_table mapper. 836 """ 837 838 sess = fixture_session() 839 840 name = "dogbert" 841 m1 = sess.query(Manager).filter(Manager.name == name).one() 842 sess.expire(m1) 843 assert m1.status == "regular manager" 844 845 name = "pointy haired boss" 846 m2 = sess.query(Manager).filter(Manager.name == name).one() 847 sess.expire(m2, ["manager_name", "golf_swing"]) 848 assert m2.golf_swing == "fore" 849 850 def test_with_polymorphic_one_future(self): 851 sess = fixture_session() 852 853 def go(): 854 wp = with_polymorphic(Person, [Engineer]) 855 eq_( 856 sess.query(wp) 857 .filter(wp.Engineer.primary_language == "java") 858 .all(), 859 self._emps_wo_relationships_fixture()[0:1], 860 ) 861 862 self.assert_sql_count(testing.db, go, 1) 863 864 def test_with_polymorphic_two_future_adhoc_wp(self): 865 """test #7262 866 867 compare to 868 test_with_polymorphic_two_future_default_wp 869 870 """ 871 872 sess = fixture_session() 873 874 def go(): 875 876 wp = with_polymorphic(Person, "*", selectable=None) 877 eq_( 878 sess.query(wp).order_by(wp.person_id).all(), 879 self._emps_wo_relationships_fixture(), 880 ) 881 882 self.assert_sql_count(testing.db, go, 1) 883 884 def test_with_polymorphic_three_future(self): 885 sess = fixture_session() 886 887 def go(): 888 wp = with_polymorphic(Person, [Engineer]) 889 890 eq_( 891 sess.query(wp).order_by(wp.person_id).all(), 892 self._emps_wo_relationships_fixture(), 893 ) 894 895 self.assert_sql_count(testing.db, go, 3) 896 897 def test_with_polymorphic_four_future(self): 898 sess = fixture_session() 899 900 def go(): 901 wp = with_polymorphic( 902 Person, Engineer, selectable=people.outerjoin(engineers) 903 ) 904 eq_( 905 sess.query(wp).order_by(wp.person_id).all(), 906 self._emps_wo_relationships_fixture(), 907 ) 908 909 self.assert_sql_count(testing.db, go, 3) 910 911 def test_with_polymorphic_five_future_override_selectable(self): 912 """test part of #7262 913 914 this is kind of a hack though, people wouldn't know to do this 915 this way. 916 917 """ 918 sess = fixture_session() 919 920 def go(): 921 # needs both [Person] and the selectable=None part 922 # TODO: why do we need [Person] and can't send []? possible 923 # bug 924 wp = with_polymorphic(Person, [Person], selectable=None) 925 926 # limit the polymorphic join down to just "Person", 927 # overriding select_table 928 eq_( 929 sess.query(wp).all(), 930 self._emps_wo_relationships_fixture(), 931 ) 932 933 self.assert_sql_count(testing.db, go, 6) 934 935 def test_with_polymorphic_six_future(self): 936 assert_raises( 937 sa_exc.InvalidRequestError, with_polymorphic, Person, [Paperwork] 938 ) 939 assert_raises( 940 sa_exc.InvalidRequestError, with_polymorphic, Engineer, [Boss] 941 ) 942 assert_raises( 943 sa_exc.InvalidRequestError, with_polymorphic, Engineer, [Person] 944 ) 945 946 def test_with_polymorphic_seven_future(self): 947 sess = fixture_session() 948 # compare to entities without related collections to prevent 949 # additional lazy SQL from firing on loaded entities 950 wp = with_polymorphic(Person, "*") 951 eq_( 952 sess.query(wp).order_by(wp.person_id).all(), 953 self._emps_wo_relationships_fixture(), 954 ) 955 956 def test_relationship_to_polymorphic_one(self): 957 expected = self._company_with_emps_machines_fixture() 958 sess = fixture_session() 959 960 def go(): 961 # test load Companies with lazy load to 'employees' 962 eq_(sess.query(Company).all(), expected) 963 964 count = {"": 10, "Polymorphic": 5}.get(self.select_type, 6) 965 self.assert_sql_count(testing.db, go, count) 966 967 def test_relationship_to_polymorphic_two(self): 968 expected = self._company_with_emps_machines_fixture() 969 sess = fixture_session() 970 971 def go(): 972 # with #2438, of_type() is recognized. This 973 # overrides the with_polymorphic of the mapper 974 # and we get a consistent 3 queries now. 975 eq_( 976 sess.query(Company) 977 .options( 978 joinedload(Company.employees.of_type(Engineer)).joinedload( 979 Engineer.machines 980 ) 981 ) 982 .all(), 983 expected, 984 ) 985 986 # in the old case, we would get this 987 # count = {'':7, 'Polymorphic':1}.get(self.select_type, 2) 988 989 # query one is company->Person/Engineer->Machines 990 # query two is managers + boss for row #3 991 # query three is managers for row #4 992 count = 3 993 self.assert_sql_count(testing.db, go, count) 994 995 def test_relationship_to_polymorphic_three(self): 996 expected = self._company_with_emps_machines_fixture() 997 sess = fixture_session() 998 999 sess = fixture_session() 1000 1001 def go(): 1002 eq_( 1003 sess.query(Company) 1004 .options( 1005 subqueryload( 1006 Company.employees.of_type(Engineer) 1007 ).subqueryload(Engineer.machines) 1008 ) 1009 .all(), 1010 expected, 1011 ) 1012 1013 # the old case where subqueryload_all 1014 # didn't work with of_tyoe 1015 # count = { '':8, 'Joins':4, 'Unions':4, 'Polymorphic':3, 1016 # 'AliasedJoins':4}[self.select_type] 1017 1018 # query one is company->Person/Engineer->Machines 1019 # query two is Person/Engineer subq 1020 # query three is Machines subq 1021 # (however this test can't tell if the Q was a 1022 # lazyload or subqload ...) 1023 # query four is managers + boss for row #3 1024 # query five is managers for row #4 1025 count = 5 1026 self.assert_sql_count(testing.db, go, count) 1027 1028 def test_joinedload_on_subclass(self): 1029 sess = fixture_session() 1030 expected = [ 1031 Engineer( 1032 name="dilbert", 1033 engineer_name="dilbert", 1034 primary_language="java", 1035 status="regular engineer", 1036 machines=[ 1037 Machine(name="IBM ThinkPad"), 1038 Machine(name="IPhone"), 1039 ], 1040 ) 1041 ] 1042 1043 def go(): 1044 # test load People with joinedload to engineers + machines 1045 wp = with_polymorphic(Person, "*") 1046 eq_( 1047 sess.query(wp) 1048 .options(joinedload(wp.Engineer.machines)) 1049 .filter(wp.name == "dilbert") 1050 .all(), 1051 expected, 1052 ) 1053 1054 self.assert_sql_count(testing.db, go, 1) 1055 1056 def test_subqueryload_on_subclass(self): 1057 sess = fixture_session() 1058 expected = [ 1059 Engineer( 1060 name="dilbert", 1061 engineer_name="dilbert", 1062 primary_language="java", 1063 status="regular engineer", 1064 machines=[ 1065 Machine(name="IBM ThinkPad"), 1066 Machine(name="IPhone"), 1067 ], 1068 ) 1069 ] 1070 1071 def go(): 1072 wp = with_polymorphic(Person, "*") 1073 eq_( 1074 sess.query(wp) 1075 .options(subqueryload(wp.Engineer.machines)) 1076 .filter(wp.name == "dilbert") 1077 .all(), 1078 expected, 1079 ) 1080 1081 self.assert_sql_count(testing.db, go, 2) 1082 1083 def test_query_subclass_join_to_base_relationship(self): 1084 sess = fixture_session() 1085 # non-polymorphic 1086 eq_(sess.query(Engineer).join(Person.paperwork).all(), [e1, e2, e3]) 1087 1088 def test_join_to_subclass_manual_alias(self): 1089 sess = fixture_session() 1090 1091 target = aliased(Engineer, people.join(engineers)) 1092 eq_( 1093 sess.query(Company) 1094 .join(Company.employees.of_type(target)) 1095 .filter(target.primary_language == "java") 1096 .all(), 1097 [c1], 1098 ) 1099 1100 def test_join_to_subclass_one(self): 1101 sess = fixture_session() 1102 eq_( 1103 sess.query(Company) 1104 .select_from(companies.join(people).join(engineers)) 1105 .filter(Engineer.primary_language == "java") 1106 .all(), 1107 [c1], 1108 ) 1109 1110 def test_join_to_subclass_three(self): 1111 sess = fixture_session() 1112 ealias = aliased(Engineer) 1113 eq_( 1114 sess.query(Company) 1115 .join(ealias, Company.employees) 1116 .filter(ealias.primary_language == "java") 1117 .all(), 1118 [c1], 1119 ) 1120 1121 def test_join_to_subclass_six(self): 1122 sess = fixture_session() 1123 1124 eq_( 1125 sess.query(Company) 1126 .join(Company.employees.of_type(Engineer)) 1127 .join(Engineer.machines) 1128 .all(), 1129 [c1, c2], 1130 ) 1131 1132 def test_join_to_subclass_six_point_five(self): 1133 sess = fixture_session() 1134 1135 q = ( 1136 sess.query(Company) 1137 .join(Company.employees.of_type(Engineer)) 1138 .join(Engineer.machines) 1139 .filter(Engineer.name == "dilbert") 1140 ) 1141 self.assert_compile( 1142 q, 1143 "SELECT companies.company_id AS companies_company_id, " 1144 "companies.name AS companies_name FROM companies JOIN " 1145 "(people JOIN engineers ON people.person_id = " 1146 "engineers.person_id) ON " 1147 "companies.company_id = people.company_id " 1148 "JOIN machines ON engineers.person_id = machines.engineer_id " 1149 "WHERE people.name = :name_1", 1150 ) 1151 eq_( 1152 q.all(), 1153 [c1], 1154 ) 1155 1156 def test_join_to_subclass_eight(self): 1157 sess = fixture_session() 1158 eq_(sess.query(Person).join(Engineer.machines).all(), [e1, e2, e3]) 1159 1160 def test_join_to_subclass_nine(self): 1161 sess = fixture_session() 1162 eq_( 1163 sess.query(Company) 1164 .select_from(companies.join(people).join(engineers)) 1165 .filter(Engineer.primary_language == "java") 1166 .all(), 1167 [c1], 1168 ) 1169 1170 def test_join_to_subclass_ten(self): 1171 sess = fixture_session() 1172 eq_( 1173 sess.query(Company) 1174 .join(Company.employees) 1175 .filter(Engineer.primary_language == "java") 1176 .all(), 1177 [c1], 1178 ) 1179 1180 def test_join_to_subclass_eleven(self): 1181 sess = fixture_session() 1182 eq_( 1183 sess.query(Company) 1184 .select_from(companies.join(people).join(engineers)) 1185 .filter(Engineer.primary_language == "java") 1186 .all(), 1187 [c1], 1188 ) 1189 1190 def test_join_to_subclass_twelve(self): 1191 sess = fixture_session() 1192 eq_(sess.query(Person).join(Engineer.machines).all(), [e1, e2, e3]) 1193 1194 def test_join_to_subclass_thirteen(self): 1195 sess = fixture_session() 1196 eq_( 1197 sess.query(Person) 1198 .join(Engineer.machines) 1199 .filter(Machine.name.ilike("%ibm%")) 1200 .all(), 1201 [e1, e3], 1202 ) 1203 1204 def test_join_to_subclass_fourteen(self): 1205 sess = fixture_session() 1206 eq_( 1207 sess.query(Company) 1208 .join(Company.employees) 1209 .join(Engineer.machines) 1210 .all(), 1211 [c1, c2], 1212 ) 1213 1214 def test_join_to_subclass_fifteen(self): 1215 sess = fixture_session() 1216 eq_( 1217 sess.query(Company) 1218 .join(Company.employees) 1219 .join(Engineer.machines) 1220 .filter(Machine.name.ilike("%thinkpad%")) 1221 .all(), 1222 [c1], 1223 ) 1224 1225 def test_join_to_subclass_sixteen(self): 1226 sess = fixture_session() 1227 # non-polymorphic 1228 eq_(sess.query(Engineer).join(Engineer.machines).all(), [e1, e2, e3]) 1229 1230 def test_join_to_subclass_seventeen(self): 1231 sess = fixture_session() 1232 eq_( 1233 sess.query(Engineer) 1234 .join(Engineer.machines) 1235 .filter(Machine.name.ilike("%ibm%")) 1236 .all(), 1237 [e1, e3], 1238 ) 1239 1240 def test_join_and_thru_polymorphic_nonaliased_one(self): 1241 sess = fixture_session() 1242 eq_( 1243 sess.query(Company) 1244 .join(Company.employees) 1245 .join(Person.paperwork.and_(Paperwork.description.like("%#2%"))) 1246 .all(), 1247 [c1], 1248 ) 1249 1250 def test_join_and_thru_polymorphic_aliased_one(self): 1251 sess = fixture_session() 1252 ea = aliased(Person) 1253 pa = aliased(Paperwork) 1254 eq_( 1255 sess.query(Company) 1256 .join(ea, Company.employees) 1257 .join(pa, ea.paperwork.and_(pa.description.like("%#2%"))) 1258 .all(), 1259 [c1], 1260 ) 1261 1262 def test_join_through_polymorphic_nonaliased_one(self): 1263 sess = fixture_session() 1264 eq_( 1265 sess.query(Company) 1266 .join(Company.employees) 1267 .join(Person.paperwork) 1268 .filter(Paperwork.description.like("%#2%")) 1269 .all(), 1270 [c1], 1271 ) 1272 1273 def test_join_through_polymorphic_nonaliased_two(self): 1274 sess = fixture_session() 1275 eq_( 1276 sess.query(Company) 1277 .join(Company.employees) 1278 .join(Person.paperwork) 1279 .filter(Paperwork.description.like("%#%")) 1280 .all(), 1281 [c1, c2], 1282 ) 1283 1284 def test_join_through_polymorphic_nonaliased_three(self): 1285 sess = fixture_session() 1286 eq_( 1287 sess.query(Company) 1288 .join(Company.employees) 1289 .join(Person.paperwork) 1290 .filter(Person.name.in_(["dilbert", "vlad"])) 1291 .filter(Paperwork.description.like("%#2%")) 1292 .all(), 1293 [c1], 1294 ) 1295 1296 def test_join_through_polymorphic_nonaliased_four(self): 1297 sess = fixture_session() 1298 eq_( 1299 sess.query(Company) 1300 .join(Company.employees) 1301 .join(Person.paperwork) 1302 .filter(Person.name.in_(["dilbert", "vlad"])) 1303 .filter(Paperwork.description.like("%#%")) 1304 .all(), 1305 [c1, c2], 1306 ) 1307 1308 def test_join_through_polymorphic_nonaliased_five(self): 1309 sess = fixture_session() 1310 eq_( 1311 sess.query(Company) 1312 .join(Company.employees) 1313 .filter(Person.name.in_(["dilbert", "vlad"])) 1314 .join(Person.paperwork) 1315 .filter(Paperwork.description.like("%#2%")) 1316 .all(), 1317 [c1], 1318 ) 1319 1320 def test_join_through_polymorphic_nonaliased_six(self): 1321 sess = fixture_session() 1322 eq_( 1323 sess.query(Company) 1324 .join(Company.employees) 1325 .filter(Person.name.in_(["dilbert", "vlad"])) 1326 .join(Person.paperwork) 1327 .filter(Paperwork.description.like("%#%")) 1328 .all(), 1329 [c1, c2], 1330 ) 1331 1332 def test_join_through_polymorphic_aliased_one(self): 1333 sess = fixture_session() 1334 ea = aliased(Person) 1335 pa = aliased(Paperwork) 1336 eq_( 1337 sess.query(Company) 1338 .join(ea, Company.employees) 1339 .join(pa, ea.paperwork) 1340 .filter(pa.description.like("%#2%")) 1341 .all(), 1342 [c1], 1343 ) 1344 1345 def test_join_through_polymorphic_aliased_two(self): 1346 sess = fixture_session() 1347 ea = aliased(Person) 1348 pa = aliased(Paperwork) 1349 eq_( 1350 sess.query(Company) 1351 .join(ea, Company.employees) 1352 .join(pa, ea.paperwork) 1353 .filter(pa.description.like("%#%")) 1354 .all(), 1355 [c1, c2], 1356 ) 1357 1358 def test_join_through_polymorphic_aliased_three(self): 1359 sess = fixture_session() 1360 ea = aliased(Person) 1361 pa = aliased(Paperwork) 1362 eq_( 1363 sess.query(Company) 1364 .join(ea, Company.employees) 1365 .join(pa, ea.paperwork) 1366 .filter(ea.name.in_(["dilbert", "vlad"])) 1367 .filter(pa.description.like("%#2%")) 1368 .all(), 1369 [c1], 1370 ) 1371 1372 def test_join_through_polymorphic_aliased_four(self): 1373 sess = fixture_session() 1374 ea = aliased(Person) 1375 pa = aliased(Paperwork) 1376 eq_( 1377 sess.query(Company) 1378 .join(ea, Company.employees) 1379 .join(pa, ea.paperwork) # we can't use "paperwork" here? 1380 .filter(ea.name.in_(["dilbert", "vlad"])) 1381 .filter(pa.description.like("%#%")) 1382 .all(), 1383 [c1, c2], 1384 ) 1385 1386 def test_join_through_polymorphic_aliased_five(self): 1387 sess = fixture_session() 1388 ea = aliased(Person) 1389 pa = aliased(Paperwork) 1390 eq_( 1391 sess.query(Company) 1392 .join(ea, Company.employees) 1393 .filter(ea.name.in_(["dilbert", "vlad"])) 1394 .join(pa, ea.paperwork) 1395 .filter(pa.description.like("%#2%")) 1396 .all(), 1397 [c1], 1398 ) 1399 1400 def test_join_through_polymorphic_aliased_six(self): 1401 sess = fixture_session() 1402 pa = aliased(Paperwork) 1403 ea = aliased(Person) 1404 eq_( 1405 sess.query(Company) 1406 .join(ea, Company.employees) 1407 .filter(ea.name.in_(["dilbert", "vlad"])) 1408 .join(pa, ea.paperwork) 1409 .filter(pa.description.like("%#%")) 1410 .all(), 1411 [c1, c2], 1412 ) 1413 1414 def test_explicit_polymorphic_join_one(self): 1415 sess = fixture_session() 1416 1417 # join from Company to Engineer; join condition formulated by 1418 # ORMJoin using regular table foreign key connections. Engineer 1419 # is expressed as "(select * people join engineers) as anon_1" 1420 # so the join is contained. 1421 eq_( 1422 sess.query(Company) 1423 .join(Engineer) 1424 .filter(Engineer.engineer_name == "vlad") 1425 .one(), 1426 c2, 1427 ) 1428 1429 def test_explicit_polymorphic_join_two(self): 1430 sess = fixture_session() 1431 1432 # same, using explicit join condition. Query.join() must 1433 # adapt the on clause here to match the subquery wrapped around 1434 # "people join engineers". 1435 eq_( 1436 sess.query(Company) 1437 .join(Engineer, Company.company_id == Engineer.company_id) 1438 .filter(Engineer.engineer_name == "vlad") 1439 .one(), 1440 c2, 1441 ) 1442 1443 def test_filter_on_baseclass(self): 1444 sess = fixture_session() 1445 eq_(sess.query(Person).order_by(Person.person_id).all(), all_employees) 1446 eq_( 1447 sess.query(Person).order_by(Person.person_id).first(), 1448 all_employees[0], 1449 ) 1450 eq_( 1451 sess.query(Person) 1452 .order_by(Person.person_id) 1453 .filter(Person.person_id == e2.person_id) 1454 .one(), 1455 e2, 1456 ) 1457 1458 def test_from_alias(self): 1459 sess = fixture_session() 1460 palias = aliased(Person) 1461 eq_( 1462 sess.query(palias) 1463 .order_by(palias.person_id) 1464 .filter(palias.name.in_(["dilbert", "wally"])) 1465 .all(), 1466 [e1, e2], 1467 ) 1468 1469 def test_self_referential_one(self): 1470 sess = fixture_session() 1471 palias = aliased(Person) 1472 expected = [(m1, e1), (m1, e2), (m1, b1)] 1473 1474 eq_( 1475 sess.query(Person, palias) 1476 .filter(Person.company_id == palias.company_id) 1477 .filter(Person.name == "dogbert") 1478 .filter(Person.person_id > palias.person_id) 1479 .order_by(Person.person_id, palias.person_id) 1480 .all(), 1481 expected, 1482 ) 1483 1484 def test_self_referential_two(self): 1485 1486 sess = fixture_session() 1487 palias = aliased(Person) 1488 expected = [(m1, e1), (m1, e2), (m1, b1)] 1489 1490 with testing.expect_deprecated(r"The Query.from_self\(\) method"): 1491 eq_( 1492 sess.query(Person, palias) 1493 .filter(Person.company_id == palias.company_id) 1494 .filter(Person.name == "dogbert") 1495 .filter(Person.person_id > palias.person_id) 1496 .from_self() 1497 .order_by(Person.person_id, palias.person_id) 1498 .all(), 1499 expected, 1500 ) 1501 1502 def test_self_referential_two_point_five(self): 1503 """Using two aliases, the above case works.""" 1504 sess = fixture_session() 1505 palias = aliased(Person) 1506 palias2 = aliased(Person) 1507 1508 expected = [(m1, e1), (m1, e2), (m1, b1)] 1509 1510 with testing.expect_deprecated(r"The Query.from_self\(\) method"): 1511 eq_( 1512 sess.query(palias, palias2) 1513 .filter(palias.company_id == palias2.company_id) 1514 .filter(palias.name == "dogbert") 1515 .filter(palias.person_id > palias2.person_id) 1516 .from_self() 1517 .order_by(palias.person_id, palias2.person_id) 1518 .all(), 1519 expected, 1520 ) 1521 1522 def test_self_referential_two_future(self): 1523 # TODO: this is the SECOND test *EVER* of an aliased class of 1524 # an aliased class. 1525 sess = fixture_session(future=True) 1526 expected = [(m1, e1), (m1, e2), (m1, b1)] 1527 1528 # not aliasing the first class 1529 p1 = Person 1530 p2 = aliased(Person) 1531 stmt = ( 1532 select(p1, p2) 1533 .filter(p1.company_id == p2.company_id) 1534 .filter(p1.name == "dogbert") 1535 .filter(p1.person_id > p2.person_id) 1536 ) 1537 1538 subq = stmt.subquery() 1539 1540 pa1 = aliased(p1, subq) 1541 pa2 = aliased(p2, subq) 1542 1543 stmt2 = select(pa1, pa2).order_by(pa1.person_id, pa2.person_id) 1544 1545 eq_( 1546 sess.execute(stmt2).unique().all(), 1547 expected, 1548 ) 1549 1550 def test_self_referential_two_point_five_future(self): 1551 1552 # TODO: this is the first test *EVER* of an aliased class of 1553 # an aliased class. we should add many more tests for this. 1554 # new case added in Id810f485c5f7ed971529489b84694e02a3356d6d 1555 sess = fixture_session(future=True) 1556 expected = [(m1, e1), (m1, e2), (m1, b1)] 1557 1558 # aliasing the first class 1559 p1 = aliased(Person) 1560 p2 = aliased(Person) 1561 stmt = ( 1562 select(p1, p2) 1563 .filter(p1.company_id == p2.company_id) 1564 .filter(p1.name == "dogbert") 1565 .filter(p1.person_id > p2.person_id) 1566 ) 1567 subq = stmt.subquery() 1568 1569 pa1 = aliased(p1, subq) 1570 pa2 = aliased(p2, subq) 1571 1572 stmt2 = select(pa1, pa2).order_by(pa1.person_id, pa2.person_id) 1573 1574 eq_( 1575 sess.execute(stmt2).unique().all(), 1576 expected, 1577 ) 1578 1579 def test_nesting_queries(self): 1580 # query.statement places a flag "no_adapt" on the returned 1581 # statement. This prevents the polymorphic adaptation in the 1582 # second "filter" from hitting it, which would pollute the 1583 # subquery and usually results in recursion overflow errors 1584 # within the adaption. 1585 sess = fixture_session() 1586 subq = ( 1587 sess.query(engineers.c.person_id) 1588 .filter(Engineer.primary_language == "java") 1589 .statement.scalar_subquery() 1590 ) 1591 1592 eq_(sess.query(Person).filter(Person.person_id.in_(subq)).one(), e1) 1593 1594 def test_mixed_entities_one(self): 1595 sess = fixture_session() 1596 1597 expected = [ 1598 ( 1599 Engineer( 1600 status="regular engineer", 1601 engineer_name="dilbert", 1602 name="dilbert", 1603 company_id=1, 1604 primary_language="java", 1605 person_id=1, 1606 type="engineer", 1607 ), 1608 "MegaCorp, Inc.", 1609 ), 1610 ( 1611 Engineer( 1612 status="regular engineer", 1613 engineer_name="wally", 1614 name="wally", 1615 company_id=1, 1616 primary_language="c++", 1617 person_id=2, 1618 type="engineer", 1619 ), 1620 "MegaCorp, Inc.", 1621 ), 1622 ( 1623 Engineer( 1624 status="elbonian engineer", 1625 engineer_name="vlad", 1626 name="vlad", 1627 company_id=2, 1628 primary_language="cobol", 1629 person_id=5, 1630 type="engineer", 1631 ), 1632 "Elbonia, Inc.", 1633 ), 1634 ] 1635 eq_( 1636 sess.query(Engineer, Company.name) 1637 .join(Company.employees) 1638 .order_by(Person.person_id) 1639 .filter(Person.type == "engineer") 1640 .all(), 1641 expected, 1642 ) 1643 1644 def _join_to_poly_wp_one(self, sess): 1645 wp = with_polymorphic(self.classes.Person, "*") 1646 return ( 1647 sess.query(wp.name, self.classes.Company.name) 1648 .join(self.classes.Company.employees.of_type(wp)) 1649 .order_by(wp.person_id) 1650 ) 1651 1652 def _join_to_poly_wp_two(self, sess): 1653 wp = with_polymorphic(self.classes.Person, "*", aliased=True) 1654 return ( 1655 sess.query(wp.name, self.classes.Company.name) 1656 .join(self.classes.Company.employees.of_type(wp)) 1657 .order_by(wp.person_id) 1658 ) 1659 1660 def _join_to_poly_wp_three(self, sess): 1661 wp = with_polymorphic( 1662 self.classes.Person, "*", aliased=True, flat=True 1663 ) 1664 return ( 1665 sess.query(wp.name, self.classes.Company.name) 1666 .join(self.classes.Company.employees.of_type(wp)) 1667 .order_by(wp.person_id) 1668 ) 1669 1670 @testing.combinations( 1671 lambda self, sess: ( 1672 sess.query(self.classes.Person.name, self.classes.Company.name) 1673 .join(self.classes.Company.employees) 1674 .order_by(self.classes.Person.person_id) 1675 ), 1676 _join_to_poly_wp_one, 1677 _join_to_poly_wp_two, 1678 _join_to_poly_wp_three, 1679 ) 1680 def test_mixed_entities_join_to_poly(self, q): 1681 sess = fixture_session() 1682 expected = [ 1683 ("dilbert", "MegaCorp, Inc."), 1684 ("wally", "MegaCorp, Inc."), 1685 ("pointy haired boss", "MegaCorp, Inc."), 1686 ("dogbert", "MegaCorp, Inc."), 1687 ("vlad", "Elbonia, Inc."), 1688 ] 1689 eq_( 1690 q(self, sess).all(), 1691 expected, 1692 ) 1693 1694 def test_mixed_entities_two(self): 1695 sess = fixture_session() 1696 expected = [ 1697 ("java", "MegaCorp, Inc."), 1698 ("cobol", "Elbonia, Inc."), 1699 ("c++", "MegaCorp, Inc."), 1700 ] 1701 eq_( 1702 sess.query(Engineer.primary_language, Company.name) 1703 .join(Company.employees) 1704 .filter(Person.type == "engineer") 1705 .order_by(desc(Engineer.primary_language)) 1706 .all(), 1707 expected, 1708 ) 1709 1710 def test_mixed_entities_three(self): 1711 sess = fixture_session() 1712 palias = aliased(Person) 1713 expected = [ 1714 ( 1715 Engineer( 1716 status="elbonian engineer", 1717 engineer_name="vlad", 1718 name="vlad", 1719 primary_language="cobol", 1720 ), 1721 "Elbonia, Inc.", 1722 Engineer( 1723 status="regular engineer", 1724 engineer_name="dilbert", 1725 name="dilbert", 1726 company_id=1, 1727 primary_language="java", 1728 person_id=1, 1729 type="engineer", 1730 ), 1731 ) 1732 ] 1733 eq_( 1734 sess.query(Person, Company.name, palias) 1735 .join(Company.employees) 1736 .filter(Company.name == "Elbonia, Inc.") 1737 .filter(palias.name == "dilbert") 1738 .filter(palias.person_id != Person.person_id) 1739 .all(), 1740 expected, 1741 ) 1742 1743 def test_mixed_entities_four(self): 1744 sess = fixture_session() 1745 palias = aliased(Person) 1746 expected = [ 1747 ( 1748 Engineer( 1749 status="regular engineer", 1750 engineer_name="dilbert", 1751 name="dilbert", 1752 company_id=1, 1753 primary_language="java", 1754 person_id=1, 1755 type="engineer", 1756 ), 1757 "Elbonia, Inc.", 1758 Engineer( 1759 status="elbonian engineer", 1760 engineer_name="vlad", 1761 name="vlad", 1762 primary_language="cobol", 1763 ), 1764 ) 1765 ] 1766 1767 eq_( 1768 sess.query(palias, Company.name, Person) 1769 .select_from(join(palias, Company, true())) 1770 .join(Company.employees) 1771 .filter(Company.name == "Elbonia, Inc.") 1772 .filter(palias.name == "dilbert") 1773 .all(), 1774 expected, 1775 ) 1776 1777 def test_mixed_entities_five(self): 1778 sess = fixture_session() 1779 palias = aliased(Person) 1780 expected = [("vlad", "Elbonia, Inc.", "dilbert")] 1781 eq_( 1782 sess.query(Person.name, Company.name, palias.name) 1783 .join(Company.employees) 1784 .filter(Company.name == "Elbonia, Inc.") 1785 .filter(palias.name == "dilbert") 1786 .filter(palias.company_id != Person.company_id) 1787 .all(), 1788 expected, 1789 ) 1790 1791 def test_mixed_entities_six(self): 1792 sess = fixture_session() 1793 palias = aliased(Person) 1794 expected = [ 1795 ("manager", "dogbert", "engineer", "dilbert"), 1796 ("manager", "dogbert", "engineer", "wally"), 1797 ("manager", "dogbert", "boss", "pointy haired boss"), 1798 ] 1799 eq_( 1800 sess.query(Person.type, Person.name, palias.type, palias.name) 1801 .filter(Person.company_id == palias.company_id) 1802 .filter(Person.name == "dogbert") 1803 .filter(Person.person_id > palias.person_id) 1804 .order_by(Person.person_id, palias.person_id) 1805 .all(), 1806 expected, 1807 ) 1808 1809 def test_mixed_entities_seven(self): 1810 sess = fixture_session() 1811 expected = [ 1812 ("dilbert", "tps report #1"), 1813 ("dilbert", "tps report #2"), 1814 ("dogbert", "review #2"), 1815 ("dogbert", "review #3"), 1816 ("pointy haired boss", "review #1"), 1817 ("vlad", "elbonian missive #3"), 1818 ("wally", "tps report #3"), 1819 ("wally", "tps report #4"), 1820 ] 1821 eq_( 1822 sess.query(Person.name, Paperwork.description) 1823 .filter(Person.person_id == Paperwork.person_id) 1824 .order_by(Person.name, Paperwork.description) 1825 .all(), 1826 expected, 1827 ) 1828 1829 def test_mixed_entities_eight(self): 1830 sess = fixture_session() 1831 eq_( 1832 sess.query(func.count(Person.person_id)) 1833 .filter(Engineer.primary_language == "java") 1834 .all(), 1835 [(1,)], 1836 ) 1837 1838 def test_mixed_entities_nine(self): 1839 sess = fixture_session() 1840 expected = [("Elbonia, Inc.", 1), ("MegaCorp, Inc.", 4)] 1841 eq_( 1842 sess.query(Company.name, func.count(Person.person_id)) 1843 .filter(Company.company_id == Person.company_id) 1844 .group_by(Company.name) 1845 .order_by(Company.name) 1846 .all(), 1847 expected, 1848 ) 1849 1850 def test_mixed_entities_ten(self): 1851 sess = fixture_session() 1852 expected = [("Elbonia, Inc.", 1), ("MegaCorp, Inc.", 4)] 1853 eq_( 1854 sess.query(Company.name, func.count(Person.person_id)) 1855 .join(Company.employees) 1856 .group_by(Company.name) 1857 .order_by(Company.name) 1858 .all(), 1859 expected, 1860 ) 1861 1862 # def test_mixed_entities(self): 1863 # sess = fixture_session() 1864 # TODO: I think raise error on these for now. different 1865 # inheritance/loading schemes have different results here, 1866 # all incorrect 1867 # 1868 # eq_( 1869 # sess.query(Person.name, Engineer.primary_language).all(), 1870 # []) 1871 1872 # def test_mixed_entities(self): 1873 # sess = fixture_session() 1874 # eq_(sess.query( 1875 # Person.name, 1876 # Engineer.primary_language, 1877 # Manager.manager_name) 1878 # .all(), 1879 # []) 1880 1881 def test_mixed_entities_eleven(self): 1882 sess = fixture_session() 1883 expected = [("java",), ("c++",), ("cobol",)] 1884 eq_( 1885 sess.query(Engineer.primary_language) 1886 .filter(Person.type == "engineer") 1887 .all(), 1888 expected, 1889 ) 1890 1891 def test_mixed_entities_twelve(self): 1892 sess = fixture_session() 1893 expected = [("vlad", "Elbonia, Inc.")] 1894 eq_( 1895 sess.query(Person.name, Company.name) 1896 .join(Company.employees) 1897 .filter(Company.name == "Elbonia, Inc.") 1898 .all(), 1899 expected, 1900 ) 1901 1902 def test_mixed_entities_thirteen(self): 1903 sess = fixture_session() 1904 expected = [("pointy haired boss", "fore")] 1905 eq_(sess.query(Boss.name, Boss.golf_swing).all(), expected) 1906 1907 def test_mixed_entities_fourteen(self): 1908 sess = fixture_session() 1909 expected = [("dilbert", "java"), ("wally", "c++"), ("vlad", "cobol")] 1910 eq_( 1911 sess.query(Engineer.name, Engineer.primary_language).all(), 1912 expected, 1913 ) 1914 1915 def test_mixed_entities_fifteen(self): 1916 sess = fixture_session() 1917 1918 expected = [ 1919 ( 1920 "Elbonia, Inc.", 1921 Engineer( 1922 status="elbonian engineer", 1923 engineer_name="vlad", 1924 name="vlad", 1925 primary_language="cobol", 1926 ), 1927 ) 1928 ] 1929 eq_( 1930 sess.query(Company.name, Person) 1931 .join(Company.employees) 1932 .filter(Company.name == "Elbonia, Inc.") 1933 .all(), 1934 expected, 1935 ) 1936 1937 def test_mixed_entities_sixteen(self): 1938 sess = fixture_session() 1939 expected = [ 1940 ( 1941 Engineer( 1942 status="elbonian engineer", 1943 engineer_name="vlad", 1944 name="vlad", 1945 primary_language="cobol", 1946 ), 1947 "Elbonia, Inc.", 1948 ) 1949 ] 1950 eq_( 1951 sess.query(Person, Company.name) 1952 .join(Company.employees) 1953 .filter(Company.name == "Elbonia, Inc.") 1954 .all(), 1955 expected, 1956 ) 1957 1958 def test_mixed_entities_seventeen(self): 1959 sess = fixture_session() 1960 expected = [("pointy haired boss",), ("dogbert",)] 1961 eq_(sess.query(Manager.name).all(), expected) 1962 1963 def test_mixed_entities_eighteen(self): 1964 sess = fixture_session() 1965 expected = [("pointy haired boss foo",), ("dogbert foo",)] 1966 eq_(sess.query(Manager.name + " foo").all(), expected) 1967 1968 def test_mixed_entities_nineteen(self): 1969 sess = fixture_session() 1970 row = ( 1971 sess.query(Engineer.name, Engineer.primary_language) 1972 .filter(Engineer.name == "dilbert") 1973 .first() 1974 ) 1975 assert row.name == "dilbert" 1976 assert row.primary_language == "java" 1977 1978 def test_correlation_one(self): 1979 sess = fixture_session() 1980 1981 # this for a long time did not work with PolymorphicAliased and 1982 # PolymorphicUnions, which was due to the no_replacement_traverse 1983 # annotation added to query.statement which then went into 1984 # scalar_subquery(). this is removed as of :ticket:`4304` so now 1985 # works. 1986 eq_( 1987 sess.query(Person.name) 1988 .filter( 1989 sess.query(Company.name) 1990 .filter(Company.company_id == Person.company_id) 1991 .correlate(Person) 1992 .scalar_subquery() 1993 == "Elbonia, Inc." 1994 ) 1995 .all(), 1996 [(e3.name,)], 1997 ) 1998 1999 def test_correlation_two(self): 2000 sess = fixture_session() 2001 2002 paliased = aliased(Person) 2003 2004 eq_( 2005 sess.query(paliased.name) 2006 .filter( 2007 sess.query(Company.name) 2008 .filter(Company.company_id == paliased.company_id) 2009 .correlate(paliased) 2010 .scalar_subquery() 2011 == "Elbonia, Inc." 2012 ) 2013 .all(), 2014 [(e3.name,)], 2015 ) 2016 2017 def test_correlation_three(self): 2018 sess = fixture_session() 2019 2020 paliased = aliased(Person, flat=True) 2021 2022 eq_( 2023 sess.query(paliased.name) 2024 .filter( 2025 sess.query(Company.name) 2026 .filter(Company.company_id == paliased.company_id) 2027 .correlate(paliased) 2028 .scalar_subquery() 2029 == "Elbonia, Inc." 2030 ) 2031 .all(), 2032 [(e3.name,)], 2033 ) 2034 2035 2036class PolymorphicTest(_PolymorphicTestBase, _Polymorphic): 2037 def test_primary_eager_aliasing_three_dont_reset_selectable(self): 2038 """test now related to #7262 2039 2040 See test_primary_eager_aliasing_three_reset_selectable for 2041 the reset selectable version. 2042 2043 """ 2044 # assert the JOINs don't over JOIN 2045 2046 sess = fixture_session() 2047 2048 # selectable default is False 2049 wp = with_polymorphic(Person, "*") 2050 2051 def go(): 2052 eq_( 2053 sess.query(wp) 2054 .order_by(wp.person_id) 2055 .options(joinedload(wp.Engineer.machines))[1:3], 2056 all_employees[1:3], 2057 ) 2058 2059 self.assert_sql_count(testing.db, go, 3) 2060 2061 eq_( 2062 sess.scalar( 2063 select(func.count("*")).select_from( 2064 sess.query(wp) 2065 .options(joinedload(wp.Engineer.machines)) 2066 .order_by(wp.person_id) 2067 .limit(2) 2068 .offset(1) 2069 .subquery() 2070 ) 2071 ), 2072 2, 2073 ) 2074 2075 def test_with_polymorphic_two_future_default_wp(self): 2076 """test #7262 2077 2078 compare to 2079 test_with_polymorphic_two_future_adhoc_wp 2080 2081 """ 2082 sess = fixture_session() 2083 2084 def go(): 2085 2086 wp = with_polymorphic(Person, "*") 2087 eq_( 2088 sess.query(wp).order_by(wp.person_id).all(), 2089 self._emps_wo_relationships_fixture(), 2090 ) 2091 2092 self.assert_sql_count(testing.db, go, 1) 2093 2094 def test_join_to_subclass_four(self): 2095 sess = fixture_session() 2096 eq_( 2097 sess.query(Person) 2098 .select_from(people.join(engineers)) 2099 .join(Engineer.machines) 2100 .all(), 2101 [e1, e2, e3], 2102 ) 2103 2104 def test_join_to_subclass_five(self): 2105 sess = fixture_session() 2106 eq_( 2107 sess.query(Person) 2108 .select_from(people.join(engineers)) 2109 .join(Engineer.machines) 2110 .filter(Machine.name.ilike("%ibm%")) 2111 .all(), 2112 [e1, e3], 2113 ) 2114 2115 def test_correlation_w_polymorphic(self): 2116 2117 sess = fixture_session() 2118 2119 p_poly = with_polymorphic(Person, "*") 2120 2121 eq_( 2122 sess.query(p_poly.name) 2123 .filter( 2124 sess.query(Company.name) 2125 .filter(Company.company_id == p_poly.company_id) 2126 .correlate(p_poly) 2127 .scalar_subquery() 2128 == "Elbonia, Inc." 2129 ) 2130 .all(), 2131 [(e3.name,)], 2132 ) 2133 2134 def test_correlation_w_polymorphic_flat(self): 2135 2136 sess = fixture_session() 2137 2138 p_poly = with_polymorphic(Person, "*", flat=True) 2139 2140 eq_( 2141 sess.query(p_poly.name) 2142 .filter( 2143 sess.query(Company.name) 2144 .filter(Company.company_id == p_poly.company_id) 2145 .correlate(p_poly) 2146 .scalar_subquery() 2147 == "Elbonia, Inc." 2148 ) 2149 .all(), 2150 [(e3.name,)], 2151 ) 2152 2153 def test_join_to_subclass_ten(self): 2154 pass 2155 2156 def test_mixed_entities_one(self): 2157 pass 2158 2159 def test_mixed_entities_two(self): 2160 pass 2161 2162 def test_mixed_entities_eight(self): 2163 pass 2164 2165 def test_polymorphic_any_eight(self): 2166 pass 2167 2168 2169class PolymorphicPolymorphicTest( 2170 _PolymorphicTestBase, _PolymorphicPolymorphic 2171): 2172 __dialect__ = "default" 2173 2174 def test_with_polymorphic_two_future_default_wp(self): 2175 """test #7262 2176 2177 compare to 2178 test_with_polymorphic_two_future_adhoc_wp 2179 2180 """ 2181 sess = fixture_session() 2182 2183 def go(): 2184 2185 wp = with_polymorphic(Person, "*") 2186 eq_( 2187 sess.query(wp).order_by(wp.person_id).all(), 2188 self._emps_wo_relationships_fixture(), 2189 ) 2190 2191 self.assert_sql_count(testing.db, go, 1) 2192 2193 def test_aliased_not_polluted_by_join(self): 2194 # aliased(polymorphic) will normally do the old-school 2195 # "(SELECT * FROM a JOIN b ...) AS anon_1" thing. 2196 # this is the safest 2197 sess = fixture_session() 2198 palias = aliased(Person) 2199 self.assert_compile( 2200 sess.query(palias, Company.name) 2201 .order_by(palias.person_id) 2202 .join(Person, Company.employees) 2203 .filter(palias.name == "dilbert"), 2204 "SELECT anon_1.people_person_id AS anon_1_people_person_id, " 2205 "anon_1.people_company_id AS anon_1_people_company_id, " 2206 "anon_1.people_name AS anon_1_people_name, " 2207 "anon_1.people_type AS anon_1_people_type, " 2208 "anon_1.engineers_person_id AS anon_1_engineers_person_id, " 2209 "anon_1.engineers_status AS anon_1_engineers_status, " 2210 "anon_1.engineers_engineer_name AS anon_1_engineers_engineer_name, " # noqa 2211 "anon_1.engineers_primary_language AS " 2212 "anon_1_engineers_primary_language, " 2213 "anon_1.managers_person_id AS anon_1_managers_person_id, " 2214 "anon_1.managers_status AS anon_1_managers_status, " 2215 "anon_1.managers_manager_name AS anon_1_managers_manager_name, " 2216 "anon_1.boss_boss_id AS anon_1_boss_boss_id, " 2217 "anon_1.boss_golf_swing AS anon_1_boss_golf_swing, " 2218 "companies.name AS companies_name " 2219 "FROM (SELECT people.person_id AS people_person_id, " 2220 "people.company_id AS people_company_id, " 2221 "people.name AS people_name, people.type AS people_type, " 2222 "engineers.person_id AS engineers_person_id, " 2223 "engineers.status AS engineers_status, " 2224 "engineers.engineer_name AS engineers_engineer_name, " 2225 "engineers.primary_language AS engineers_primary_language, " 2226 "managers.person_id AS managers_person_id, " 2227 "managers.status AS managers_status, " 2228 "managers.manager_name AS managers_manager_name, " 2229 "boss.boss_id AS boss_boss_id, " 2230 "boss.golf_swing AS boss_golf_swing " 2231 "FROM people LEFT OUTER JOIN engineers " 2232 "ON people.person_id = engineers.person_id " 2233 "LEFT OUTER JOIN managers " 2234 "ON people.person_id = managers.person_id LEFT OUTER JOIN boss " 2235 "ON managers.person_id = boss.boss_id) AS anon_1, " 2236 "companies JOIN " 2237 "(people LEFT OUTER JOIN engineers " 2238 "ON people.person_id = engineers.person_id " 2239 "LEFT OUTER JOIN managers " 2240 "ON people.person_id = managers.person_id " 2241 "LEFT OUTER JOIN boss ON managers.person_id = boss.boss_id) " 2242 "ON companies.company_id = people.company_id " 2243 "WHERE anon_1.people_name = :people_name_1 " 2244 "ORDER BY anon_1.people_person_id", 2245 ) 2246 2247 def test_flat_aliased_w_select_from(self): 2248 sess = fixture_session() 2249 palias = aliased(Person, flat=True) 2250 self.assert_compile( 2251 sess.query(palias, Company.name) 2252 .select_from(palias) 2253 .order_by(palias.person_id) 2254 .join(Person, Company.employees) 2255 .filter(palias.name == "dilbert"), 2256 "SELECT people_1.person_id AS people_1_person_id, " 2257 "people_1.company_id AS people_1_company_id, " 2258 "people_1.name AS people_1_name, people_1.type AS people_1_type, " 2259 "engineers_1.person_id AS engineers_1_person_id, " 2260 "engineers_1.status AS engineers_1_status, " 2261 "engineers_1.engineer_name AS engineers_1_engineer_name, " 2262 "engineers_1.primary_language AS engineers_1_primary_language, " 2263 "managers_1.person_id AS managers_1_person_id, " 2264 "managers_1.status AS managers_1_status, " 2265 "managers_1.manager_name AS managers_1_manager_name, " 2266 "boss_1.boss_id AS boss_1_boss_id, " 2267 "boss_1.golf_swing AS boss_1_golf_swing, " 2268 "companies.name AS companies_name " 2269 "FROM people AS people_1 " 2270 "LEFT OUTER JOIN engineers AS engineers_1 " 2271 "ON people_1.person_id = engineers_1.person_id " 2272 "LEFT OUTER JOIN managers AS managers_1 " 2273 "ON people_1.person_id = managers_1.person_id " 2274 "LEFT OUTER JOIN boss AS boss_1 " 2275 "ON managers_1.person_id = boss_1.boss_id, " 2276 "companies JOIN (people LEFT OUTER JOIN engineers " 2277 "ON people.person_id = engineers.person_id " 2278 "LEFT OUTER JOIN managers " 2279 "ON people.person_id = managers.person_id " 2280 "LEFT OUTER JOIN boss ON managers.person_id = boss.boss_id) " 2281 "ON companies.company_id = people.company_id " 2282 "WHERE people_1.name = :name_1 ORDER BY people_1.person_id", 2283 ) 2284 2285 2286class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions): 2287 def test_with_polymorphic_two_future_default_wp(self): 2288 """test #7262 2289 2290 compare to 2291 test_with_polymorphic_two_future_adhoc_wp 2292 2293 """ 2294 sess = fixture_session() 2295 2296 def go(): 2297 2298 wp = with_polymorphic(Person, "*") 2299 eq_( 2300 sess.query(wp).order_by(wp.person_id).all(), 2301 self._emps_wo_relationships_fixture(), 2302 ) 2303 2304 self.assert_sql_count(testing.db, go, 2) 2305 2306 def test_subqueryload_on_subclass_uses_path_correctly(self): 2307 sess = fixture_session() 2308 expected = [ 2309 Engineer( 2310 name="dilbert", 2311 engineer_name="dilbert", 2312 primary_language="java", 2313 status="regular engineer", 2314 machines=[ 2315 Machine(name="IBM ThinkPad"), 2316 Machine(name="IPhone"), 2317 ], 2318 ) 2319 ] 2320 2321 with self.sql_execution_asserter(testing.db) as asserter: 2322 wp = with_polymorphic(Person, "*") 2323 eq_( 2324 sess.query(wp) 2325 .options(subqueryload(wp.Engineer.machines)) 2326 .filter(wp.name == "dilbert") 2327 .all(), 2328 expected, 2329 ) 2330 2331 asserter.assert_( 2332 CompiledSQL( 2333 "SELECT pjoin.person_id AS pjoin_person_id, " 2334 "pjoin.company_id AS pjoin_company_id, " 2335 "pjoin.name AS pjoin_name, pjoin.type AS pjoin_type, " 2336 "pjoin.status AS pjoin_status, " 2337 "pjoin.engineer_name AS pjoin_engineer_name, " 2338 "pjoin.primary_language AS pjoin_primary_language, " 2339 "pjoin.manager_name AS pjoin_manager_name " 2340 "FROM (SELECT engineers.person_id AS person_id, " 2341 "people.company_id AS company_id, people.name AS name, " 2342 "people.type AS type, engineers.status AS status, " 2343 "engineers.engineer_name AS engineer_name, " 2344 "engineers.primary_language AS primary_language, " 2345 "CAST(NULL AS VARCHAR(50)) AS manager_name " 2346 "FROM people JOIN engineers ON people.person_id = " 2347 "engineers.person_id UNION ALL SELECT managers.person_id " 2348 "AS person_id, people.company_id AS company_id, people.name " 2349 "AS name, people.type AS type, managers.status AS status, " 2350 "CAST(NULL AS VARCHAR(50)) AS engineer_name, " 2351 "CAST(NULL AS VARCHAR(50)) AS primary_language, " 2352 "managers.manager_name AS manager_name FROM people " 2353 "JOIN managers ON people.person_id = managers.person_id) " 2354 "AS pjoin WHERE pjoin.name = :name_1", 2355 params=[{"name_1": "dilbert"}], 2356 ), 2357 CompiledSQL( 2358 "SELECT machines.machine_id AS machines_machine_id, " 2359 "machines.name AS machines_name, machines.engineer_id " 2360 "AS machines_engineer_id, anon_1.pjoin_person_id AS " 2361 "anon_1_pjoin_person_id FROM " 2362 "(SELECT pjoin.person_id AS pjoin_person_id FROM " 2363 "(SELECT engineers.person_id AS person_id, people.company_id " 2364 "AS company_id, people.name AS name, " 2365 "people.type AS type, engineers.status AS status, " 2366 "engineers.engineer_name AS engineer_name, " 2367 "engineers.primary_language AS primary_language, " 2368 "CAST(NULL AS VARCHAR(50)) AS manager_name FROM people " 2369 "JOIN engineers ON people.person_id = engineers.person_id " 2370 "UNION ALL SELECT managers.person_id AS person_id, " 2371 "people.company_id AS company_id, people.name AS name, " 2372 "people.type AS type, managers.status AS status, " 2373 "CAST(NULL AS VARCHAR(50)) AS engineer_name, " 2374 "CAST(NULL AS VARCHAR(50)) AS primary_language, " 2375 "managers.manager_name AS manager_name FROM people " 2376 "JOIN managers ON people.person_id = managers.person_id) " 2377 "AS pjoin WHERE pjoin.name = :name_1) AS anon_1 JOIN " 2378 "machines ON anon_1.pjoin_person_id = machines.engineer_id " 2379 "ORDER BY machines.machine_id", 2380 params=[{"name_1": "dilbert"}], 2381 ), 2382 ) 2383 2384 2385class PolymorphicAliasedJoinsTest( 2386 _PolymorphicTestBase, _PolymorphicAliasedJoins 2387): 2388 def test_with_polymorphic_two_future_default_wp(self): 2389 """test #7262 2390 2391 compare to 2392 test_with_polymorphic_two_future_adhoc_wp 2393 2394 """ 2395 sess = fixture_session() 2396 2397 def go(): 2398 2399 wp = with_polymorphic(Person, "*") 2400 eq_( 2401 sess.query(wp).order_by(wp.person_id).all(), 2402 self._emps_wo_relationships_fixture(), 2403 ) 2404 2405 self.assert_sql_count(testing.db, go, 2) 2406 2407 2408class PolymorphicJoinsTest(_PolymorphicTestBase, _PolymorphicJoins): 2409 def test_with_polymorphic_two_future_default_wp(self): 2410 """test #7262 2411 2412 compare to 2413 test_with_polymorphic_two_future_adhoc_wp 2414 2415 """ 2416 sess = fixture_session() 2417 2418 def go(): 2419 2420 wp = with_polymorphic(Person, "*") 2421 eq_( 2422 sess.query(wp).order_by(wp.person_id).all(), 2423 self._emps_wo_relationships_fixture(), 2424 ) 2425 2426 self.assert_sql_count(testing.db, go, 2) 2427 2428 def test_having_group_by(self): 2429 sess = fixture_session() 2430 eq_( 2431 sess.query(Person.name) 2432 .group_by(Person.name) 2433 .having(Person.name == "dilbert") 2434 .all(), 2435 [("dilbert",)], 2436 ) 2437