1"""tests of joined-eager loaded attributes""" 2 3from sqlalchemy.testing import eq_, is_, is_not_ 4import sqlalchemy as sa 5from sqlalchemy import testing 6from sqlalchemy.orm import joinedload, deferred, undefer, \ 7 joinedload_all, backref, Session,\ 8 defaultload, Load, load_only 9from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, \ 10 func, text 11from sqlalchemy.testing.schema import Table, Column 12from sqlalchemy.orm import mapper, relationship, create_session, \ 13 lazyload, aliased, column_property 14from sqlalchemy.sql import operators 15from sqlalchemy.testing import assert_raises, assert_raises_message 16from sqlalchemy.testing.assertsql import CompiledSQL 17from sqlalchemy.testing import fixtures, expect_warnings 18from test.orm import _fixtures 19from sqlalchemy.util import OrderedDict as odict 20import datetime 21 22 23class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): 24 run_inserts = 'once' 25 run_deletes = None 26 __dialect__ = 'default' 27 28 def test_basic(self): 29 users, Address, addresses, User = ( 30 self.tables.users, 31 self.classes.Address, 32 self.tables.addresses, 33 self.classes.User) 34 35 mapper(User, users, properties={ 36 'addresses': relationship( 37 mapper(Address, addresses), lazy='joined', order_by=Address.id) 38 }) 39 sess = create_session() 40 q = sess.query(User) 41 42 eq_([User(id=7, addresses=[ 43 Address(id=1, email_address='jack@bean.com')])], 44 q.filter(User.id == 7).all()) 45 eq_(self.static.user_address_result, q.order_by(User.id).all()) 46 47 def test_late_compile(self): 48 User, Address, addresses, users = ( 49 self.classes.User, 50 self.classes.Address, 51 self.tables.addresses, 52 self.tables.users) 53 54 m = mapper(User, users) 55 sess = create_session() 56 sess.query(User).all() 57 m.add_property("addresses", relationship(mapper(Address, addresses))) 58 59 sess.expunge_all() 60 61 def go(): 62 eq_( 63 [User(id=7, addresses=[ 64 Address(id=1, email_address='jack@bean.com')])], 65 sess.query(User).options( 66 joinedload('addresses')).filter(User.id == 7).all() 67 ) 68 self.assert_sql_count(testing.db, go, 1) 69 70 def test_no_orphan(self): 71 """An eagerly loaded child object is not marked as an orphan""" 72 73 users, Address, addresses, User = ( 74 self.tables.users, 75 self.classes.Address, 76 self.tables.addresses, 77 self.classes.User) 78 79 mapper(User, users, properties={ 80 'addresses': relationship( 81 Address, cascade="all,delete-orphan", lazy='joined') 82 }) 83 mapper(Address, addresses) 84 85 sess = create_session() 86 user = sess.query(User).get(7) 87 assert getattr(User, 'addresses').\ 88 hasparent( 89 sa.orm.attributes.instance_state( 90 user.addresses[0]), optimistic=True) 91 assert not sa.orm.class_mapper(Address).\ 92 _is_orphan( 93 sa.orm.attributes.instance_state(user.addresses[0])) 94 95 def test_orderby(self): 96 users, Address, addresses, User = ( 97 self.tables.users, 98 self.classes.Address, 99 self.tables.addresses, 100 self.classes.User) 101 102 mapper(User, users, properties={ 103 'addresses': relationship( 104 mapper(Address, addresses), 105 lazy='joined', order_by=addresses.c.email_address), 106 }) 107 q = create_session().query(User) 108 eq_([ 109 User(id=7, addresses=[ 110 Address(id=1) 111 ]), 112 User(id=8, addresses=[ 113 Address(id=3, email_address='ed@bettyboop.com'), 114 Address(id=4, email_address='ed@lala.com'), 115 Address(id=2, email_address='ed@wood.com') 116 ]), 117 User(id=9, addresses=[ 118 Address(id=5) 119 ]), 120 User(id=10, addresses=[]) 121 ], q.order_by(User.id).all()) 122 123 def test_orderby_multi(self): 124 users, Address, addresses, User = ( 125 self.tables.users, 126 self.classes.Address, 127 self.tables.addresses, 128 self.classes.User) 129 130 mapper(User, users, properties={ 131 'addresses': relationship( 132 mapper(Address, addresses), 133 lazy='joined', 134 order_by=[addresses.c.email_address, addresses.c.id]), 135 }) 136 q = create_session().query(User) 137 eq_([ 138 User(id=7, addresses=[ 139 Address(id=1) 140 ]), 141 User(id=8, addresses=[ 142 Address(id=3, email_address='ed@bettyboop.com'), 143 Address(id=4, email_address='ed@lala.com'), 144 Address(id=2, email_address='ed@wood.com') 145 ]), 146 User(id=9, addresses=[ 147 Address(id=5) 148 ]), 149 User(id=10, addresses=[]) 150 ], q.order_by(User.id).all()) 151 152 def test_orderby_related(self): 153 """A regular mapper select on a single table can 154 order by a relationship to a second table""" 155 156 Address, addresses, users, User = (self.classes.Address, 157 self.tables.addresses, 158 self.tables.users, 159 self.classes.User) 160 161 mapper(Address, addresses) 162 mapper(User, users, properties=dict( 163 addresses=relationship( 164 Address, lazy='joined', order_by=addresses.c.id), 165 )) 166 167 q = create_session().query(User) 168 l = q.filter(User.id == Address.user_id).order_by( 169 Address.email_address).all() 170 171 eq_([ 172 User(id=8, addresses=[ 173 Address(id=2, email_address='ed@wood.com'), 174 Address(id=3, email_address='ed@bettyboop.com'), 175 Address(id=4, email_address='ed@lala.com'), 176 ]), 177 User(id=9, addresses=[ 178 Address(id=5) 179 ]), 180 User(id=7, addresses=[ 181 Address(id=1) 182 ]), 183 ], l) 184 185 def test_orderby_desc(self): 186 Address, addresses, users, User = (self.classes.Address, 187 self.tables.addresses, 188 self.tables.users, 189 self.classes.User) 190 191 mapper(Address, addresses) 192 mapper(User, users, properties=dict( 193 addresses=relationship( 194 Address, lazy='joined', 195 order_by=[sa.desc(addresses.c.email_address)]), 196 )) 197 sess = create_session() 198 eq_([ 199 User(id=7, addresses=[ 200 Address(id=1) 201 ]), 202 User(id=8, addresses=[ 203 Address(id=2, email_address='ed@wood.com'), 204 Address(id=4, email_address='ed@lala.com'), 205 Address(id=3, email_address='ed@bettyboop.com'), 206 ]), 207 User(id=9, addresses=[ 208 Address(id=5) 209 ]), 210 User(id=10, addresses=[]) 211 ], sess.query(User).order_by(User.id).all()) 212 213 def test_no_ad_hoc_orderby(self): 214 """part of #2992; make sure string label references can't 215 access an eager loader, else an eager load can corrupt the query. 216 217 """ 218 Address, addresses, users, User = (self.classes.Address, 219 self.tables.addresses, 220 self.tables.users, 221 self.classes.User) 222 223 mapper(Address, addresses) 224 mapper(User, users, properties=dict( 225 addresses=relationship( 226 Address), 227 )) 228 229 sess = create_session() 230 q = sess.query(User).\ 231 join("addresses").\ 232 options(joinedload("addresses")).\ 233 order_by("email_address") 234 235 self.assert_compile( 236 q, 237 "SELECT users.id AS users_id, users.name AS users_name, " 238 "addresses_1.id AS addresses_1_id, addresses_1.user_id AS " 239 "addresses_1_user_id, addresses_1.email_address AS " 240 "addresses_1_email_address FROM users JOIN addresses " 241 "ON users.id = addresses.user_id LEFT OUTER JOIN addresses " 242 "AS addresses_1 ON users.id = addresses_1.user_id " 243 "ORDER BY addresses.email_address" 244 ) 245 246 q = sess.query(User).options(joinedload("addresses")).\ 247 order_by("email_address") 248 249 with expect_warnings("Can't resolve label reference 'email_address'"): 250 self.assert_compile( 251 q, 252 "SELECT users.id AS users_id, users.name AS users_name, " 253 "addresses_1.id AS addresses_1_id, addresses_1.user_id AS " 254 "addresses_1_user_id, addresses_1.email_address AS " 255 "addresses_1_email_address FROM users LEFT OUTER JOIN " 256 "addresses AS addresses_1 ON users.id = addresses_1.user_id " 257 "ORDER BY email_address" 258 ) 259 260 261 262 def test_deferred_fk_col(self): 263 users, Dingaling, User, dingalings, Address, addresses = ( 264 self.tables.users, 265 self.classes.Dingaling, 266 self.classes.User, 267 self.tables.dingalings, 268 self.classes.Address, 269 self.tables.addresses) 270 271 mapper(Address, addresses, properties={ 272 'user_id': deferred(addresses.c.user_id), 273 'user': relationship(User, lazy='joined') 274 }) 275 mapper(User, users) 276 277 sess = create_session() 278 279 for q in [ 280 sess.query(Address).filter( 281 Address.id.in_([1, 4, 5]) 282 ).order_by(Address.id), 283 sess.query(Address).filter( 284 Address.id.in_([1, 4, 5]) 285 ).order_by(Address.id).limit(3) 286 ]: 287 sess.expunge_all() 288 eq_(q.all(), 289 [Address(id=1, user=User(id=7)), 290 Address(id=4, user=User(id=8)), 291 Address(id=5, user=User(id=9))] 292 ) 293 294 sess.expunge_all() 295 a = sess.query(Address).filter(Address.id == 1).all()[0] 296 297 # 1.0 change! we don't automatically undefer user_id here. 298 # if the user wants a column undeferred, add the option. 299 def go(): 300 eq_(a.user_id, 7) 301 # self.assert_sql_count(testing.db, go, 0) 302 self.assert_sql_count(testing.db, go, 1) 303 304 sess.expunge_all() 305 a = sess.query(Address).filter(Address.id == 1).first() 306 307 def go(): 308 eq_(a.user_id, 7) 309 # same, 1.0 doesn't check these 310 # self.assert_sql_count(testing.db, go, 0) 311 self.assert_sql_count(testing.db, go, 1) 312 313 # do the mapping in reverse 314 # (we would have just used an "addresses" backref but the test 315 # fixtures then require the whole backref to be set up, lazy loaders 316 # trigger, etc.) 317 sa.orm.clear_mappers() 318 319 mapper(Address, addresses, properties={ 320 'user_id': deferred(addresses.c.user_id), 321 }) 322 mapper(User, users, properties={ 323 'addresses': relationship(Address, lazy='joined')}) 324 325 for q in [ 326 sess.query(User).filter(User.id == 7), 327 sess.query(User).filter(User.id == 7).limit(1) 328 ]: 329 sess.expunge_all() 330 eq_(q.all(), 331 [User(id=7, addresses=[Address(id=1)])] 332 ) 333 334 sess.expunge_all() 335 u = sess.query(User).get(7) 336 337 def go(): 338 eq_(u.addresses[0].user_id, 7) 339 # assert that the eager loader didn't have to affect 'user_id' here 340 # and that its still deferred 341 self.assert_sql_count(testing.db, go, 1) 342 343 sa.orm.clear_mappers() 344 345 mapper(User, users, properties={ 346 'addresses': relationship(Address, lazy='joined', 347 order_by=addresses.c.id)}) 348 mapper(Address, addresses, properties={ 349 'user_id': deferred(addresses.c.user_id), 350 'dingalings': relationship(Dingaling, lazy='joined')}) 351 mapper(Dingaling, dingalings, properties={ 352 'address_id': deferred(dingalings.c.address_id)}) 353 sess.expunge_all() 354 355 def go(): 356 u = sess.query(User).get(8) 357 eq_(User(id=8, 358 addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), 359 Address(id=3), 360 Address(id=4)]), 361 u) 362 self.assert_sql_count(testing.db, go, 1) 363 364 def test_options_pathing(self): 365 users, Keyword, orders, items, order_items, \ 366 Order, Item, User, keywords, item_keywords = ( 367 self.tables.users, 368 self.classes.Keyword, 369 self.tables.orders, 370 self.tables.items, 371 self.tables.order_items, 372 self.classes.Order, 373 self.classes.Item, 374 self.classes.User, 375 self.tables.keywords, 376 self.tables.item_keywords) 377 378 mapper(User, users, properties={ 379 'orders': relationship(Order, order_by=orders.c.id), # o2m, m2o 380 }) 381 mapper(Order, orders, properties={ 382 'items': relationship( 383 Item, 384 secondary=order_items, order_by=items.c.id), # m2m 385 }) 386 mapper(Item, items, properties={ 387 'keywords': relationship(Keyword, 388 secondary=item_keywords, 389 order_by=keywords.c.id) # m2m 390 }) 391 mapper(Keyword, keywords) 392 393 for opt, count in [ 394 (( 395 joinedload(User.orders, Order.items), 396 ), 10), 397 ((joinedload("orders.items"), ), 10), 398 (( 399 joinedload(User.orders, ), 400 joinedload(User.orders, Order.items), 401 joinedload(User.orders, Order.items, Item.keywords), 402 ), 1), 403 (( 404 joinedload(User.orders, Order.items, Item.keywords), 405 ), 10), 406 (( 407 joinedload(User.orders, Order.items), 408 joinedload(User.orders, Order.items, Item.keywords), 409 ), 5), 410 ]: 411 sess = create_session() 412 413 def go(): 414 eq_( 415 sess.query(User).options(*opt).order_by(User.id).all(), 416 self.static.user_item_keyword_result 417 ) 418 self.assert_sql_count(testing.db, go, count) 419 420 def test_disable_dynamic(self): 421 """test no joined option on a dynamic.""" 422 423 users, Address, addresses, User = ( 424 self.tables.users, 425 self.classes.Address, 426 self.tables.addresses, 427 self.classes.User) 428 429 mapper(User, users, properties={ 430 'addresses': relationship(Address, lazy="dynamic") 431 }) 432 mapper(Address, addresses) 433 sess = create_session() 434 assert_raises_message( 435 sa.exc.InvalidRequestError, 436 "User.addresses' does not support object " 437 "population - eager loading cannot be applied.", 438 sess.query(User).options(joinedload(User.addresses)).first, 439 ) 440 441 def test_many_to_many(self): 442 keywords, items, item_keywords, Keyword, Item = ( 443 self.tables.keywords, 444 self.tables.items, 445 self.tables.item_keywords, 446 self.classes.Keyword, 447 self.classes.Item) 448 449 mapper(Keyword, keywords) 450 mapper(Item, items, properties=dict( 451 keywords=relationship(Keyword, secondary=item_keywords, 452 lazy='joined', order_by=keywords.c.id))) 453 454 q = create_session().query(Item).order_by(Item.id) 455 456 def go(): 457 eq_(self.static.item_keyword_result, q.all()) 458 self.assert_sql_count(testing.db, go, 1) 459 460 def go(): 461 eq_(self.static.item_keyword_result[0:2], 462 q.join('keywords').filter(Keyword.name == 'red').all()) 463 self.assert_sql_count(testing.db, go, 1) 464 465 def go(): 466 eq_(self.static.item_keyword_result[0:2], 467 (q.join('keywords', aliased=True). 468 filter(Keyword.name == 'red')).all()) 469 self.assert_sql_count(testing.db, go, 1) 470 471 def test_eager_option(self): 472 keywords, items, item_keywords, Keyword, Item = ( 473 self.tables.keywords, 474 self.tables.items, 475 self.tables.item_keywords, 476 self.classes.Keyword, 477 self.classes.Item) 478 479 mapper(Keyword, keywords) 480 mapper(Item, items, properties=dict( 481 keywords=relationship( 482 Keyword, secondary=item_keywords, lazy='select', 483 order_by=keywords.c.id))) 484 q = create_session().query(Item) 485 486 def go(): 487 eq_(self.static.item_keyword_result[0:2], 488 (q.options( 489 joinedload('keywords') 490 ).join('keywords'). 491 filter(keywords.c.name == 'red')).order_by(Item.id).all()) 492 493 self.assert_sql_count(testing.db, go, 1) 494 495 def test_cyclical(self): 496 """A circular eager relationship breaks the cycle with a lazy loader""" 497 498 Address, addresses, users, User = (self.classes.Address, 499 self.tables.addresses, 500 self.tables.users, 501 self.classes.User) 502 503 mapper(Address, addresses) 504 mapper(User, users, properties=dict( 505 addresses=relationship( 506 Address, lazy='joined', 507 backref=sa.orm.backref('user', lazy='joined'), 508 order_by=Address.id) 509 )) 510 eq_(sa.orm.class_mapper(User).get_property('addresses').lazy, 'joined') 511 eq_(sa.orm.class_mapper(Address).get_property('user').lazy, 'joined') 512 513 sess = create_session() 514 eq_( 515 self.static.user_address_result, 516 sess.query(User).order_by(User.id).all()) 517 518 def test_double(self): 519 """Eager loading with two relationships simultaneously, 520 from the same table, using aliases.""" 521 522 users, orders, User, Address, Order, addresses = ( 523 self.tables.users, 524 self.tables.orders, 525 self.classes.User, 526 self.classes.Address, 527 self.classes.Order, 528 self.tables.addresses) 529 530 openorders = sa.alias(orders, 'openorders') 531 closedorders = sa.alias(orders, 'closedorders') 532 533 mapper(Address, addresses) 534 mapper(Order, orders) 535 536 open_mapper = mapper(Order, openorders, non_primary=True) 537 closed_mapper = mapper(Order, closedorders, non_primary=True) 538 539 mapper(User, users, properties=dict( 540 addresses=relationship( 541 Address, lazy='joined', order_by=addresses.c.id), 542 open_orders=relationship( 543 open_mapper, 544 primaryjoin=sa.and_(openorders.c.isopen == 1, 545 users.c.id == openorders.c.user_id), 546 lazy='joined', order_by=openorders.c.id), 547 closed_orders=relationship( 548 closed_mapper, 549 primaryjoin=sa.and_(closedorders.c.isopen == 0, 550 users.c.id == closedorders.c.user_id), 551 lazy='joined', order_by=closedorders.c.id))) 552 553 q = create_session().query(User).order_by(User.id) 554 555 def go(): 556 eq_([ 557 User( 558 id=7, 559 addresses=[Address(id=1)], 560 open_orders=[Order(id=3)], 561 closed_orders=[Order(id=1), Order(id=5)] 562 ), 563 User( 564 id=8, 565 addresses=[Address(id=2), Address(id=3), Address(id=4)], 566 open_orders=[], 567 closed_orders=[] 568 ), 569 User( 570 id=9, 571 addresses=[Address(id=5)], 572 open_orders=[Order(id=4)], 573 closed_orders=[Order(id=2)] 574 ), 575 User(id=10) 576 577 ], q.all()) 578 self.assert_sql_count(testing.db, go, 1) 579 580 def test_double_same_mappers(self): 581 """Eager loading with two relationships simulatneously, 582 from the same table, using aliases.""" 583 584 addresses, items, order_items, orders, \ 585 Item, User, Address, Order, users = ( 586 self.tables.addresses, 587 self.tables.items, 588 self.tables.order_items, 589 self.tables.orders, 590 self.classes.Item, 591 self.classes.User, 592 self.classes.Address, 593 self.classes.Order, 594 self.tables.users) 595 596 mapper(Address, addresses) 597 mapper(Order, orders, properties={ 598 'items': relationship(Item, secondary=order_items, lazy='joined', 599 order_by=items.c.id)}) 600 mapper(Item, items) 601 mapper(User, users, properties=dict( 602 addresses=relationship( 603 Address, lazy='joined', order_by=addresses.c.id), 604 open_orders=relationship( 605 Order, 606 primaryjoin=sa.and_(orders.c.isopen == 1, 607 users.c.id == orders.c.user_id), 608 lazy='joined', order_by=orders.c.id), 609 closed_orders=relationship( 610 Order, 611 primaryjoin=sa.and_(orders.c.isopen == 0, 612 users.c.id == orders.c.user_id), 613 lazy='joined', order_by=orders.c.id))) 614 q = create_session().query(User).order_by(User.id) 615 616 def go(): 617 eq_([ 618 User(id=7, 619 addresses=[ 620 Address(id=1)], 621 open_orders=[Order(id=3, 622 items=[ 623 Item(id=3), 624 Item(id=4), 625 Item(id=5)])], 626 closed_orders=[Order(id=1, 627 items=[ 628 Item(id=1), 629 Item(id=2), 630 Item(id=3)]), 631 Order(id=5, 632 items=[ 633 Item(id=5)])]), 634 User(id=8, 635 addresses=[ 636 Address(id=2), 637 Address(id=3), 638 Address(id=4)], 639 open_orders=[], 640 closed_orders=[]), 641 User(id=9, 642 addresses=[ 643 Address(id=5)], 644 open_orders=[ 645 Order(id=4, 646 items=[ 647 Item(id=1), 648 Item(id=5)])], 649 closed_orders=[ 650 Order(id=2, 651 items=[ 652 Item(id=1), 653 Item(id=2), 654 Item(id=3)])]), 655 User(id=10) 656 ], q.all()) 657 self.assert_sql_count(testing.db, go, 1) 658 659 def test_no_false_hits(self): 660 """Eager loaders don't interpret main table columns as 661 part of their eager load.""" 662 663 addresses, orders, User, Address, Order, users = ( 664 self.tables.addresses, 665 self.tables.orders, 666 self.classes.User, 667 self.classes.Address, 668 self.classes.Order, 669 self.tables.users) 670 671 mapper(User, users, properties={ 672 'addresses': relationship(Address, lazy='joined'), 673 'orders': relationship(Order, lazy='joined') 674 }) 675 mapper(Address, addresses) 676 mapper(Order, orders) 677 678 self.allusers = create_session().query(User).all() 679 680 # using a textual select, the columns will be 'id' and 'name'. the 681 # eager loaders have aliases which should not hit on those columns, 682 # they should be required to locate only their aliased/fully table 683 # qualified column name. 684 noeagers = create_session().query(User).\ 685 from_statement(text("select * from users")).all() 686 assert 'orders' not in noeagers[0].__dict__ 687 assert 'addresses' not in noeagers[0].__dict__ 688 689 def test_limit(self): 690 """Limit operations combined with lazy-load relationships.""" 691 692 users, items, order_items, orders, Item, \ 693 User, Address, Order, addresses = ( 694 self.tables.users, 695 self.tables.items, 696 self.tables.order_items, 697 self.tables.orders, 698 self.classes.Item, 699 self.classes.User, 700 self.classes.Address, 701 self.classes.Order, 702 self.tables.addresses) 703 704 mapper(Item, items) 705 mapper(Order, orders, properties={ 706 'items': relationship(Item, secondary=order_items, lazy='joined', 707 order_by=items.c.id) 708 }) 709 mapper(User, users, properties={ 710 'addresses': relationship( 711 mapper(Address, addresses), 712 lazy='joined', order_by=addresses.c.id), 713 'orders': relationship(Order, lazy='select', order_by=orders.c.id) 714 }) 715 716 sess = create_session() 717 q = sess.query(User) 718 719 l = q.order_by(User.id).limit(2).offset(1).all() 720 eq_(self.static.user_all_result[1:3], l) 721 722 def test_distinct(self): 723 Address, addresses, users, User = (self.classes.Address, 724 self.tables.addresses, 725 self.tables.users, 726 self.classes.User) 727 728 # this is an involved 3x union of the users table to get a lot of rows. 729 # then see if the "distinct" works its way out. you actually get 730 # the same result with or without the distinct, just via less or 731 # more rows. 732 u2 = users.alias('u2') 733 s = sa.union_all( 734 u2.select(use_labels=True), u2.select(use_labels=True), 735 u2.select(use_labels=True)).alias('u') 736 737 mapper(User, users, properties={ 738 'addresses': relationship( 739 mapper(Address, addresses), 740 lazy='joined', order_by=addresses.c.id), 741 }) 742 743 sess = create_session() 744 q = sess.query(User) 745 746 def go(): 747 l = q.filter(s.c.u2_id == User.id).distinct().\ 748 order_by(User.id).all() 749 eq_(self.static.user_address_result, l) 750 self.assert_sql_count(testing.db, go, 1) 751 752 def test_limit_2(self): 753 keywords, items, item_keywords, Keyword, Item = ( 754 self.tables.keywords, 755 self.tables.items, 756 self.tables.item_keywords, 757 self.classes.Keyword, 758 self.classes.Item) 759 760 mapper(Keyword, keywords) 761 mapper(Item, items, properties=dict( 762 keywords=relationship( 763 Keyword, secondary=item_keywords, 764 lazy='joined', order_by=[keywords.c.id]), 765 )) 766 767 sess = create_session() 768 q = sess.query(Item) 769 l = q.filter((Item.description == 'item 2') | 770 (Item.description == 'item 5') | 771 (Item.description == 'item 3')).\ 772 order_by(Item.id).limit(2).all() 773 774 eq_(self.static.item_keyword_result[1:3], l) 775 776 def test_limit_3(self): 777 """test that the ORDER BY is propagated from the inner 778 select to the outer select, when using the 779 'wrapped' select statement resulting from the combination of 780 eager loading and limit/offset clauses.""" 781 782 addresses, items, order_items, orders, \ 783 Item, User, Address, Order, users = ( 784 self.tables.addresses, 785 self.tables.items, 786 self.tables.order_items, 787 self.tables.orders, 788 self.classes.Item, 789 self.classes.User, 790 self.classes.Address, 791 self.classes.Order, 792 self.tables.users) 793 794 mapper(Item, items) 795 mapper(Order, orders, properties=dict( 796 items=relationship(Item, secondary=order_items, lazy='joined') 797 )) 798 799 mapper(Address, addresses) 800 mapper(User, users, properties=dict( 801 addresses=relationship( 802 Address, lazy='joined', order_by=addresses.c.id), 803 orders=relationship(Order, lazy='joined', order_by=orders.c.id), 804 )) 805 sess = create_session() 806 807 q = sess.query(User) 808 809 if not testing.against('mssql'): 810 l = q.join('orders').order_by( 811 Order.user_id.desc()).limit(2).offset(1) 812 eq_([ 813 User(id=9, 814 orders=[Order(id=2), Order(id=4)], 815 addresses=[Address(id=5)] 816 ), 817 User(id=7, 818 orders=[Order(id=1), Order(id=3), Order(id=5)], 819 addresses=[Address(id=1)] 820 ) 821 ], l.all()) 822 823 l = q.join('addresses').order_by( 824 Address.email_address.desc()).limit(1).offset(0) 825 eq_([ 826 User(id=7, 827 orders=[Order(id=1), Order(id=3), Order(id=5)], 828 addresses=[Address(id=1)] 829 ) 830 ], l.all()) 831 832 def test_limit_4(self): 833 User, Order, addresses, users, orders = (self.classes.User, 834 self.classes.Order, 835 self.tables.addresses, 836 self.tables.users, 837 self.tables.orders) 838 839 # tests the LIMIT/OFFSET aliasing on a mapper 840 # against a select. original issue from ticket #904 841 sel = sa.select([users, addresses.c.email_address], 842 users.c.id == addresses.c.user_id).alias('useralias') 843 mapper(User, sel, properties={ 844 'orders': relationship( 845 Order, primaryjoin=sel.c.id == orders.c.user_id, 846 lazy='joined', order_by=orders.c.id) 847 }) 848 mapper(Order, orders) 849 850 sess = create_session() 851 eq_(sess.query(User).first(), 852 User(name='jack', orders=[ 853 Order( 854 address_id=1, 855 description='order 1', 856 isopen=0, 857 user_id=7, 858 id=1), 859 Order( 860 address_id=1, 861 description='order 3', 862 isopen=1, 863 user_id=7, 864 id=3), 865 Order( 866 address_id=None, description='order 5', isopen=0, 867 user_id=7, id=5)], 868 email_address='jack@bean.com', id=7) 869 ) 870 871 def test_useget_cancels_eager(self): 872 """test that a one to many lazyload cancels the unnecessary 873 eager many-to-one join on the other side.""" 874 875 users, Address, addresses, User = ( 876 self.tables.users, 877 self.classes.Address, 878 self.tables.addresses, 879 self.classes.User) 880 881 mapper(User, users) 882 mapper(Address, addresses, properties={ 883 'user': relationship(User, lazy='joined', backref='addresses') 884 }) 885 886 sess = create_session() 887 u1 = sess.query(User).filter(User.id == 8).one() 888 889 def go(): 890 eq_(u1.addresses[0].user, u1) 891 self.assert_sql_execution( 892 testing.db, go, 893 CompiledSQL( 894 "SELECT addresses.id AS addresses_id, addresses.user_id AS " 895 "addresses_user_id, addresses.email_address AS " 896 "addresses_email_address FROM addresses WHERE :param_1 = " 897 "addresses.user_id", 898 {'param_1': 8}) 899 ) 900 901 def test_manytoone_limit(self): 902 """test that the subquery wrapping only occurs with 903 limit/offset and m2m or o2m joins present.""" 904 905 users, items, order_items, Order, Item, User, \ 906 Address, orders, addresses = ( 907 self.tables.users, 908 self.tables.items, 909 self.tables.order_items, 910 self.classes.Order, 911 self.classes.Item, 912 self.classes.User, 913 self.classes.Address, 914 self.tables.orders, 915 self.tables.addresses) 916 917 mapper(User, users, properties=odict( 918 orders=relationship(Order, backref='user') 919 )) 920 mapper(Order, orders, properties=odict([ 921 ('items', relationship(Item, secondary=order_items, 922 backref='orders')), 923 ('address', relationship(Address)) 924 ])) 925 mapper(Address, addresses) 926 mapper(Item, items) 927 928 sess = create_session() 929 930 self.assert_compile( 931 sess.query(User).options(joinedload(User.orders)).limit(10), 932 "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name " 933 "AS anon_1_users_name, orders_1.id AS orders_1_id, " 934 "orders_1.user_id AS orders_1_user_id, orders_1.address_id " 935 "AS orders_1_address_id, orders_1.description AS " 936 "orders_1_description, orders_1.isopen AS orders_1_isopen " 937 "FROM (SELECT users.id AS users_id, users.name AS users_name " 938 "FROM users " 939 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS " 940 "orders_1 ON anon_1.users_id = orders_1.user_id", 941 {'param_1': 10} 942 ) 943 944 self.assert_compile( 945 sess.query(Order).options(joinedload(Order.user)).limit(10), 946 "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " 947 "orders.address_id AS " 948 "orders_address_id, orders.description AS orders_description, " 949 "orders.isopen AS orders_isopen, " 950 "users_1.id AS users_1_id, users_1.name AS users_1_name " 951 "FROM orders LEFT OUTER JOIN users AS " 952 "users_1 ON users_1.id = orders.user_id LIMIT :param_1", 953 {'param_1': 10} 954 ) 955 956 self.assert_compile( 957 sess.query(Order).options( 958 joinedload(Order.user, innerjoin=True)).limit(10), 959 "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " 960 "orders.address_id AS " 961 "orders_address_id, orders.description AS orders_description, " 962 "orders.isopen AS orders_isopen, " 963 "users_1.id AS users_1_id, users_1.name AS users_1_name " 964 "FROM orders JOIN users AS " 965 "users_1 ON users_1.id = orders.user_id LIMIT :param_1", 966 {'param_1': 10} 967 ) 968 969 self.assert_compile( 970 sess.query(User).options( 971 joinedload_all("orders.address")).limit(10), 972 "SELECT anon_1.users_id AS anon_1_users_id, " 973 "anon_1.users_name AS anon_1_users_name, " 974 "addresses_1.id AS addresses_1_id, " 975 "addresses_1.user_id AS addresses_1_user_id, " 976 "addresses_1.email_address AS addresses_1_email_address, " 977 "orders_1.id AS orders_1_id, " 978 "orders_1.user_id AS orders_1_user_id, " 979 "orders_1.address_id AS orders_1_address_id, " 980 "orders_1.description AS orders_1_description, " 981 "orders_1.isopen AS orders_1_isopen FROM " 982 "(SELECT users.id AS users_id, users.name AS users_name " 983 "FROM users LIMIT :param_1) AS anon_1 " 984 "LEFT OUTER JOIN orders AS orders_1 " 985 "ON anon_1.users_id = orders_1.user_id LEFT OUTER JOIN " 986 "addresses AS addresses_1 ON addresses_1.id = orders_1.address_id", 987 {'param_1': 10} 988 ) 989 990 self.assert_compile( 991 sess.query(User).options(joinedload_all("orders.items"), 992 joinedload("orders.address")), 993 "SELECT users.id AS users_id, users.name AS users_name, " 994 "items_1.id AS items_1_id, " 995 "items_1.description AS items_1_description, " 996 "addresses_1.id AS addresses_1_id, " 997 "addresses_1.user_id AS addresses_1_user_id, " 998 "addresses_1.email_address AS " 999 "addresses_1_email_address, orders_1.id AS orders_1_id, " 1000 "orders_1.user_id AS " 1001 "orders_1_user_id, orders_1.address_id AS orders_1_address_id, " 1002 "orders_1.description " 1003 "AS orders_1_description, orders_1.isopen AS orders_1_isopen " 1004 "FROM users LEFT OUTER JOIN orders AS orders_1 " 1005 "ON users.id = orders_1.user_id " 1006 "LEFT OUTER JOIN (order_items AS order_items_1 " 1007 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1008 "ON orders_1.id = order_items_1.order_id " 1009 "LEFT OUTER JOIN addresses AS addresses_1 " 1010 "ON addresses_1.id = orders_1.address_id" 1011 ) 1012 1013 self.assert_compile( 1014 sess.query(User).options( 1015 joinedload("orders"), 1016 joinedload( 1017 "orders.address", 1018 innerjoin=True)).limit(10), 1019 "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name " 1020 "AS anon_1_users_name, addresses_1.id AS addresses_1_id, " 1021 "addresses_1.user_id AS addresses_1_user_id, " 1022 "addresses_1.email_address AS addresses_1_email_address, " 1023 "orders_1.id AS orders_1_id, orders_1.user_id AS " 1024 "orders_1_user_id, orders_1.address_id AS orders_1_address_id, " 1025 "orders_1.description AS orders_1_description, " 1026 "orders_1.isopen AS orders_1_isopen " 1027 "FROM (SELECT users.id AS users_id, users.name AS users_name " 1028 "FROM users" 1029 " LIMIT :param_1) AS anon_1 LEFT OUTER JOIN " 1030 "(orders AS orders_1 JOIN addresses AS addresses_1 " 1031 "ON addresses_1.id = orders_1.address_id) ON " 1032 "anon_1.users_id = orders_1.user_id", 1033 {'param_1': 10} 1034 ) 1035 1036 self.assert_compile( 1037 sess.query(User).options( 1038 joinedload("orders", innerjoin=True), 1039 joinedload("orders.address", innerjoin=True)).limit(10), 1040 "SELECT anon_1.users_id AS anon_1_users_id, " 1041 "anon_1.users_name AS anon_1_users_name, " 1042 "addresses_1.id AS addresses_1_id, " 1043 "addresses_1.user_id AS addresses_1_user_id, " 1044 "addresses_1.email_address AS addresses_1_email_address, " 1045 "orders_1.id AS orders_1_id, " 1046 "orders_1.user_id AS orders_1_user_id, " 1047 "orders_1.address_id AS orders_1_address_id, " 1048 "orders_1.description AS orders_1_description, " 1049 "orders_1.isopen AS orders_1_isopen " 1050 "FROM (SELECT users.id AS users_id, users.name AS users_name " 1051 "FROM users " 1052 "LIMIT :param_1) AS anon_1 JOIN orders " 1053 "AS orders_1 ON anon_1.users_id = " 1054 "orders_1.user_id JOIN addresses AS addresses_1 " 1055 "ON addresses_1.id = orders_1.address_id", 1056 {'param_1': 10} 1057 ) 1058 1059 def test_one_to_many_scalar(self): 1060 Address, addresses, users, User = (self.classes.Address, 1061 self.tables.addresses, 1062 self.tables.users, 1063 self.classes.User) 1064 1065 mapper(User, users, properties=dict( 1066 address=relationship(mapper(Address, addresses), 1067 lazy='joined', uselist=False) 1068 )) 1069 q = create_session().query(User) 1070 1071 def go(): 1072 l = q.filter(users.c.id == 7).all() 1073 eq_([User(id=7, address=Address(id=1))], l) 1074 self.assert_sql_count(testing.db, go, 1) 1075 1076 def test_one_to_many_scalar_subq_wrapping(self): 1077 Address, addresses, users, User = (self.classes.Address, 1078 self.tables.addresses, 1079 self.tables.users, 1080 self.classes.User) 1081 1082 mapper(User, users, properties=dict( 1083 address=relationship(mapper(Address, addresses), 1084 lazy='joined', uselist=False) 1085 )) 1086 q = create_session().query(User) 1087 q = q.filter(users.c.id == 7).limit(1) 1088 1089 self.assert_compile( 1090 q, 1091 "SELECT users.id AS users_id, users.name AS users_name, " 1092 "addresses_1.id AS addresses_1_id, " 1093 "addresses_1.user_id AS addresses_1_user_id, " 1094 "addresses_1.email_address AS addresses_1_email_address " 1095 "FROM users LEFT OUTER JOIN addresses AS addresses_1 " 1096 "ON users.id = addresses_1.user_id " 1097 "WHERE users.id = :id_1 " 1098 "LIMIT :param_1", 1099 checkparams={'id_1': 7, 'param_1': 1} 1100 ) 1101 1102 def test_many_to_one(self): 1103 users, Address, addresses, User = ( 1104 self.tables.users, 1105 self.classes.Address, 1106 self.tables.addresses, 1107 self.classes.User) 1108 1109 mapper(Address, addresses, properties=dict( 1110 user=relationship(mapper(User, users), lazy='joined') 1111 )) 1112 sess = create_session() 1113 q = sess.query(Address) 1114 1115 def go(): 1116 a = q.filter(addresses.c.id == 1).one() 1117 is_not_(a.user, None) 1118 u1 = sess.query(User).get(7) 1119 is_(a.user, u1) 1120 self.assert_sql_count(testing.db, go, 1) 1121 1122 def test_many_to_one_null(self): 1123 """test that a many-to-one eager load which loads None does 1124 not later trigger a lazy load. 1125 1126 """ 1127 1128 Order, Address, addresses, orders = (self.classes.Order, 1129 self.classes.Address, 1130 self.tables.addresses, 1131 self.tables.orders) 1132 1133 # use a primaryjoin intended to defeat SA's usage of 1134 # query.get() for a many-to-one lazyload 1135 mapper(Order, orders, properties=dict( 1136 address=relationship( 1137 mapper(Address, addresses), 1138 primaryjoin=and_( 1139 addresses.c.id == orders.c.address_id, 1140 addresses.c.email_address != None 1141 ), 1142 1143 lazy='joined') 1144 )) 1145 sess = create_session() 1146 1147 def go(): 1148 o1 = sess.query(Order).options( 1149 lazyload('address')).filter( 1150 Order.id == 5).one() 1151 eq_(o1.address, None) 1152 self.assert_sql_count(testing.db, go, 2) 1153 1154 sess.expunge_all() 1155 1156 def go(): 1157 o1 = sess.query(Order).filter(Order.id == 5).one() 1158 eq_(o1.address, None) 1159 self.assert_sql_count(testing.db, go, 1) 1160 1161 def test_one_and_many(self): 1162 """tests eager load for a parent object with a child object that 1163 contains a many-to-many relationship to a third object.""" 1164 1165 users, items, order_items, orders, Item, User, Order = ( 1166 self.tables.users, 1167 self.tables.items, 1168 self.tables.order_items, 1169 self.tables.orders, 1170 self.classes.Item, 1171 self.classes.User, 1172 self.classes.Order) 1173 1174 mapper(User, users, properties={ 1175 'orders': relationship(Order, lazy='joined', order_by=orders.c.id) 1176 }) 1177 mapper(Item, items) 1178 mapper(Order, orders, properties=dict( 1179 items=relationship( 1180 Item, 1181 secondary=order_items, 1182 lazy='joined', 1183 order_by=items.c.id) 1184 )) 1185 1186 q = create_session().query(User) 1187 1188 l = q.filter(text("users.id in (7, 8, 9)")).order_by(text("users.id")) 1189 1190 def go(): 1191 eq_(self.static.user_order_result[0:3], l.all()) 1192 self.assert_sql_count(testing.db, go, 1) 1193 1194 def test_double_with_aggregate(self): 1195 User, users, orders, Order = (self.classes.User, 1196 self.tables.users, 1197 self.tables.orders, 1198 self.classes.Order) 1199 1200 max_orders_by_user = sa.select([ 1201 sa.func.max(orders.c.id).label('order_id')], 1202 group_by=[orders.c.user_id] 1203 ).alias('max_orders_by_user') 1204 1205 max_orders = orders.select( 1206 orders.c.id == max_orders_by_user.c.order_id).\ 1207 alias('max_orders') 1208 1209 mapper(Order, orders) 1210 mapper(User, users, properties={ 1211 'orders': relationship(Order, backref='user', lazy='joined', 1212 order_by=orders.c.id), 1213 'max_order': relationship( 1214 mapper(Order, max_orders, non_primary=True), 1215 lazy='joined', uselist=False) 1216 }) 1217 1218 q = create_session().query(User) 1219 1220 def go(): 1221 eq_([ 1222 User(id=7, orders=[ 1223 Order(id=1), 1224 Order(id=3), 1225 Order(id=5), 1226 ], 1227 max_order=Order(id=5) 1228 ), 1229 User(id=8, orders=[]), 1230 User(id=9, orders=[Order(id=2), Order(id=4)], 1231 max_order=Order(id=4) 1232 ), 1233 User(id=10), 1234 ], q.order_by(User.id).all()) 1235 self.assert_sql_count(testing.db, go, 1) 1236 1237 def test_uselist_false_warning(self): 1238 """test that multiple rows received by a 1239 uselist=False raises a warning.""" 1240 1241 User, users, orders, Order = (self.classes.User, 1242 self.tables.users, 1243 self.tables.orders, 1244 self.classes.Order) 1245 1246 mapper(User, users, properties={ 1247 'order': relationship(Order, uselist=False) 1248 }) 1249 mapper(Order, orders) 1250 s = create_session() 1251 assert_raises(sa.exc.SAWarning, 1252 s.query(User).options(joinedload(User.order)).all) 1253 1254 def test_wide(self): 1255 users, items, order_items, Order, Item, \ 1256 User, Address, orders, addresses = ( 1257 self.tables.users, 1258 self.tables.items, 1259 self.tables.order_items, 1260 self.classes.Order, 1261 self.classes.Item, 1262 self.classes.User, 1263 self.classes.Address, 1264 self.tables.orders, 1265 self.tables.addresses) 1266 1267 mapper( 1268 Order, orders, properties={ 1269 'items': relationship( 1270 Item, secondary=order_items, lazy='joined', 1271 order_by=items.c.id)}) 1272 mapper(Item, items) 1273 mapper(User, users, properties=dict( 1274 addresses=relationship( 1275 mapper( 1276 Address, 1277 addresses), 1278 lazy=False, 1279 order_by=addresses.c.id), 1280 orders=relationship(Order, lazy=False, order_by=orders.c.id), 1281 )) 1282 q = create_session().query(User) 1283 def go(): 1284 eq_(self.static.user_all_result, q.order_by(User.id).all()) 1285 self.assert_sql_count(testing.db, go, 1) 1286 1287 def test_against_select(self): 1288 """test eager loading of a mapper which is against a select""" 1289 1290 users, items, order_items, orders, Item, User, Order = ( 1291 self.tables.users, 1292 self.tables.items, 1293 self.tables.order_items, 1294 self.tables.orders, 1295 self.classes.Item, 1296 self.classes.User, 1297 self.classes.Order) 1298 1299 s = sa.select([orders], orders.c.isopen == 1).alias('openorders') 1300 1301 mapper(Order, s, properties={ 1302 'user': relationship(User, lazy='joined') 1303 }) 1304 mapper(User, users) 1305 mapper(Item, items) 1306 1307 q = create_session().query(Order) 1308 eq_([ 1309 Order(id=3, user=User(id=7)), 1310 Order(id=4, user=User(id=9)) 1311 ], q.all()) 1312 1313 q = q.select_from(s.join(order_items).join(items)).filter( 1314 ~Item.id.in_([1, 2, 5])) 1315 eq_([ 1316 Order(id=3, user=User(id=7)), 1317 ], q.all()) 1318 1319 def test_aliasing(self): 1320 """test that eager loading uses aliases to insulate the eager 1321 load from regular criterion against those tables.""" 1322 1323 Address, addresses, users, User = (self.classes.Address, 1324 self.tables.addresses, 1325 self.tables.users, 1326 self.classes.User) 1327 1328 mapper(User, users, properties=dict( 1329 addresses=relationship(mapper(Address, addresses), 1330 lazy='joined', order_by=addresses.c.id) 1331 )) 1332 q = create_session().query(User) 1333 l = q.filter(addresses.c.email_address == 'ed@lala.com').filter( 1334 Address.user_id == User.id).order_by(User.id) 1335 eq_(self.static.user_address_result[1:2], l.all()) 1336 1337 def test_inner_join(self): 1338 Address, addresses, users, User = (self.classes.Address, 1339 self.tables.addresses, 1340 self.tables.users, 1341 self.classes.User) 1342 1343 mapper(User, users, properties=dict( 1344 addresses=relationship(mapper(Address, addresses), lazy='joined', 1345 innerjoin=True, order_by=addresses.c.id) 1346 )) 1347 sess = create_session() 1348 eq_( 1349 [User(id=7, addresses=[Address(id=1)]), 1350 User(id=8, 1351 addresses=[Address(id=2, email_address='ed@wood.com'), 1352 Address(id=3, email_address='ed@bettyboop.com'), 1353 Address(id=4, email_address='ed@lala.com'), ]), 1354 User(id=9, addresses=[Address(id=5)])], sess.query(User).all() 1355 ) 1356 self.assert_compile( 1357 sess.query(User), 1358 "SELECT users.id AS users_id, users.name AS users_name, " 1359 "addresses_1.id AS addresses_1_id, " 1360 "addresses_1.user_id AS addresses_1_user_id, " 1361 "addresses_1.email_address AS addresses_1_email_address " 1362 "FROM users JOIN " 1363 "addresses AS addresses_1 ON users.id = addresses_1.user_id " 1364 "ORDER BY addresses_1.id") 1365 1366 def test_inner_join_unnested_chaining_options(self): 1367 users, items, order_items, Order, Item, User, orders = ( 1368 self.tables.users, 1369 self.tables.items, 1370 self.tables.order_items, 1371 self.classes.Order, 1372 self.classes.Item, 1373 self.classes.User, 1374 self.tables.orders) 1375 1376 mapper(User, users, properties=dict( 1377 orders=relationship(Order, innerjoin="unnested", 1378 lazy=False) 1379 )) 1380 mapper(Order, orders, properties=dict( 1381 items=relationship(Item, secondary=order_items, lazy=False, 1382 innerjoin="unnested") 1383 )) 1384 mapper(Item, items) 1385 1386 sess = create_session() 1387 self.assert_compile( 1388 sess.query(User), 1389 "SELECT users.id AS users_id, users.name AS users_name, " 1390 "items_1.id AS " 1391 "items_1_id, items_1.description AS items_1_description, " 1392 "orders_1.id AS " 1393 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1394 "orders_1.address_id AS " 1395 "orders_1_address_id, orders_1.description " 1396 "AS orders_1_description, " 1397 "orders_1.isopen AS orders_1_isopen FROM users " 1398 "JOIN orders AS orders_1 ON " 1399 "users.id = orders_1.user_id JOIN order_items AS order_items_1 " 1400 "ON orders_1.id = " 1401 "order_items_1.order_id JOIN items AS items_1 ON items_1.id = " 1402 "order_items_1.item_id" 1403 ) 1404 1405 self.assert_compile( 1406 sess.query(User).options(joinedload(User.orders, innerjoin=False)), 1407 "SELECT users.id AS users_id, users.name AS users_name, " 1408 "items_1.id AS " 1409 "items_1_id, items_1.description AS items_1_description, " 1410 "orders_1.id AS " 1411 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1412 "orders_1.address_id AS " 1413 "orders_1_address_id, orders_1.description " 1414 "AS orders_1_description, " 1415 "orders_1.isopen AS orders_1_isopen " 1416 "FROM users LEFT OUTER JOIN orders AS orders_1 " 1417 "ON users.id = orders_1.user_id " 1418 "LEFT OUTER JOIN (order_items AS order_items_1 " 1419 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1420 "ON orders_1.id = order_items_1.order_id" 1421 ) 1422 1423 self.assert_compile( 1424 sess.query(User).options( 1425 joinedload( 1426 User.orders, 1427 Order.items, 1428 innerjoin=False)), 1429 "SELECT users.id AS users_id, users.name AS users_name, " 1430 "items_1.id AS " 1431 "items_1_id, items_1.description AS items_1_description, " 1432 "orders_1.id AS " 1433 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1434 "orders_1.address_id AS " 1435 "orders_1_address_id, " 1436 "orders_1.description AS orders_1_description, " 1437 "orders_1.isopen AS orders_1_isopen " 1438 "FROM users JOIN orders AS orders_1 ON " 1439 "users.id = orders_1.user_id " 1440 "LEFT OUTER JOIN (order_items AS order_items_1 " 1441 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1442 "ON orders_1.id = order_items_1.order_id" 1443 1444 ) 1445 1446 def test_inner_join_nested_chaining_negative_options(self): 1447 users, items, order_items, Order, Item, User, orders = ( 1448 self.tables.users, 1449 self.tables.items, 1450 self.tables.order_items, 1451 self.classes.Order, 1452 self.classes.Item, 1453 self.classes.User, 1454 self.tables.orders) 1455 1456 mapper(User, users, properties=dict( 1457 orders=relationship(Order, innerjoin=True, 1458 lazy=False, order_by=orders.c.id) 1459 )) 1460 mapper(Order, orders, properties=dict( 1461 items=relationship(Item, secondary=order_items, lazy=False, 1462 innerjoin=True, order_by=items.c.id) 1463 )) 1464 mapper(Item, items) 1465 1466 sess = create_session() 1467 self.assert_compile( 1468 sess.query(User), 1469 "SELECT users.id AS users_id, users.name AS users_name, " 1470 "items_1.id AS " 1471 "items_1_id, items_1.description AS items_1_description, " 1472 "orders_1.id AS " 1473 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1474 "orders_1.address_id AS " 1475 "orders_1_address_id, orders_1.description " 1476 "AS orders_1_description, " 1477 "orders_1.isopen AS orders_1_isopen FROM users " 1478 "JOIN orders AS orders_1 ON " 1479 "users.id = orders_1.user_id JOIN order_items " 1480 "AS order_items_1 ON orders_1.id = " 1481 "order_items_1.order_id JOIN items AS items_1 ON items_1.id = " 1482 "order_items_1.item_id ORDER BY orders_1.id, items_1.id" 1483 ) 1484 1485 q = sess.query(User).options(joinedload(User.orders, innerjoin=False)) 1486 self.assert_compile( 1487 q, 1488 "SELECT users.id AS users_id, users.name AS users_name, " 1489 "items_1.id AS " 1490 "items_1_id, items_1.description AS items_1_description, " 1491 "orders_1.id AS " 1492 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1493 "orders_1.address_id AS " 1494 "orders_1_address_id, orders_1.description " 1495 "AS orders_1_description, " 1496 "orders_1.isopen AS orders_1_isopen " 1497 "FROM users LEFT OUTER JOIN " 1498 "(orders AS orders_1 JOIN order_items AS order_items_1 " 1499 "ON orders_1.id = order_items_1.order_id " 1500 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1501 "ON users.id = orders_1.user_id ORDER BY orders_1.id, items_1.id" 1502 ) 1503 1504 eq_( 1505 [ 1506 User(id=7, 1507 orders=[ 1508 Order( 1509 id=1, items=[ 1510 Item( 1511 id=1), Item( 1512 id=2), Item( 1513 id=3)]), 1514 Order( 1515 id=3, items=[ 1516 Item( 1517 id=3), Item( 1518 id=4), Item( 1519 id=5)]), 1520 Order(id=5, items=[Item(id=5)])]), 1521 User(id=8, orders=[]), 1522 User(id=9, orders=[ 1523 Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), 1524 Order(id=4, items=[Item(id=1), Item(id=5)]) 1525 ] 1526 ), 1527 User(id=10, orders=[]) 1528 ], 1529 q.order_by(User.id).all() 1530 ) 1531 1532 self.assert_compile( 1533 sess.query(User).options( 1534 joinedload( 1535 User.orders, 1536 Order.items, 1537 innerjoin=False)), 1538 "SELECT users.id AS users_id, users.name AS users_name, " 1539 "items_1.id AS " 1540 "items_1_id, items_1.description AS items_1_description, " 1541 "orders_1.id AS " 1542 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1543 "orders_1.address_id AS " 1544 "orders_1_address_id, orders_1.description AS " 1545 "orders_1_description, " 1546 "orders_1.isopen AS orders_1_isopen " 1547 "FROM users JOIN orders AS orders_1 ON users.id = " 1548 "orders_1.user_id " 1549 "LEFT OUTER JOIN (order_items AS order_items_1 " 1550 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1551 "ON orders_1.id = order_items_1.order_id ORDER BY " 1552 "orders_1.id, items_1.id" 1553 1554 ) 1555 1556 def test_inner_join_nested_chaining_positive_options(self): 1557 users, items, order_items, Order, Item, User, orders = ( 1558 self.tables.users, 1559 self.tables.items, 1560 self.tables.order_items, 1561 self.classes.Order, 1562 self.classes.Item, 1563 self.classes.User, 1564 self.tables.orders) 1565 1566 mapper(User, users, properties=dict( 1567 orders=relationship(Order, order_by=orders.c.id) 1568 )) 1569 mapper(Order, orders, properties=dict( 1570 items=relationship( 1571 Item, 1572 secondary=order_items, 1573 order_by=items.c.id) 1574 )) 1575 mapper(Item, items) 1576 1577 sess = create_session() 1578 q = sess.query(User).options( 1579 joinedload("orders", innerjoin=False). 1580 joinedload("items", innerjoin=True) 1581 ) 1582 1583 self.assert_compile( 1584 q, 1585 "SELECT users.id AS users_id, users.name AS users_name, " 1586 "items_1.id AS items_1_id, items_1.description " 1587 "AS items_1_description, " 1588 "orders_1.id AS orders_1_id, orders_1.user_id " 1589 "AS orders_1_user_id, " 1590 "orders_1.address_id AS orders_1_address_id, " 1591 "orders_1.description AS " 1592 "orders_1_description, orders_1.isopen AS orders_1_isopen " 1593 "FROM users LEFT OUTER JOIN (orders AS orders_1 " 1594 "JOIN order_items AS " 1595 "order_items_1 ON orders_1.id = order_items_1.order_id " 1596 "JOIN items AS " 1597 "items_1 ON items_1.id = order_items_1.item_id) " 1598 "ON users.id = orders_1.user_id " 1599 "ORDER BY orders_1.id, items_1.id" 1600 ) 1601 1602 eq_( 1603 [ 1604 User(id=7, 1605 orders=[ 1606 Order( 1607 id=1, items=[ 1608 Item( 1609 id=1), Item( 1610 id=2), Item( 1611 id=3)]), 1612 Order( 1613 id=3, items=[ 1614 Item( 1615 id=3), Item( 1616 id=4), Item( 1617 id=5)]), 1618 Order(id=5, items=[Item(id=5)])]), 1619 User(id=8, orders=[]), 1620 User(id=9, orders=[ 1621 Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), 1622 Order(id=4, items=[Item(id=1), Item(id=5)]) 1623 ] 1624 ), 1625 User(id=10, orders=[]) 1626 ], 1627 q.order_by(User.id).all() 1628 ) 1629 1630 def test_unnested_outerjoin_propagation_only_on_correct_path(self): 1631 # test #3131 1632 1633 User, users = self.classes.User, self.tables.users 1634 Order, orders = self.classes.Order, self.tables.orders 1635 Address, addresses = self.classes.Address, self.tables.addresses 1636 1637 mapper(User, users, properties=odict([ 1638 ('orders', relationship(Order)), 1639 ('addresses', relationship(Address)) 1640 ])) 1641 mapper(Order, orders) 1642 mapper(Address, addresses) 1643 1644 sess = create_session() 1645 q = sess.query(User).options( 1646 joinedload("orders"), 1647 joinedload("addresses", innerjoin="unnested"), 1648 ) 1649 1650 self.assert_compile( 1651 q, 1652 "SELECT users.id AS users_id, users.name AS users_name, " 1653 "orders_1.id AS orders_1_id, " 1654 "orders_1.user_id AS orders_1_user_id, " 1655 "orders_1.address_id AS orders_1_address_id, " 1656 "orders_1.description AS orders_1_description, " 1657 "orders_1.isopen AS orders_1_isopen, " 1658 "addresses_1.id AS addresses_1_id, " 1659 "addresses_1.user_id AS addresses_1_user_id, " 1660 "addresses_1.email_address AS addresses_1_email_address " 1661 "FROM users LEFT OUTER JOIN orders AS orders_1 " 1662 "ON users.id = orders_1.user_id JOIN addresses AS addresses_1 " 1663 "ON users.id = addresses_1.user_id" 1664 ) 1665 1666 def test_nested_outerjoin_propagation_only_on_correct_path(self): 1667 # test #3131 1668 1669 User, users = self.classes.User, self.tables.users 1670 Order, orders = self.classes.Order, self.tables.orders 1671 Address, addresses = self.classes.Address, self.tables.addresses 1672 1673 mapper(User, users, properties=odict([ 1674 ('orders', relationship(Order)), 1675 ('addresses', relationship(Address)) 1676 ])) 1677 mapper(Order, orders) 1678 mapper(Address, addresses) 1679 1680 sess = create_session() 1681 q = sess.query(User).options( 1682 joinedload("orders"), 1683 joinedload("addresses", innerjoin=True), 1684 ) 1685 1686 self.assert_compile( 1687 q, 1688 "SELECT users.id AS users_id, users.name AS users_name, " 1689 "orders_1.id AS orders_1_id, " 1690 "orders_1.user_id AS orders_1_user_id, " 1691 "orders_1.address_id AS orders_1_address_id, " 1692 "orders_1.description AS orders_1_description, " 1693 "orders_1.isopen AS orders_1_isopen, " 1694 "addresses_1.id AS addresses_1_id, " 1695 "addresses_1.user_id AS addresses_1_user_id, " 1696 "addresses_1.email_address AS addresses_1_email_address " 1697 "FROM users LEFT OUTER JOIN orders AS orders_1 " 1698 "ON users.id = orders_1.user_id JOIN addresses AS addresses_1 " 1699 "ON users.id = addresses_1.user_id" 1700 ) 1701 1702 1703 def test_catch_the_right_target(self): 1704 # test eager join chaining to the "nested" join on the left, 1705 # a new feature as of [ticket:2369] 1706 1707 users, Keyword, orders, items, order_items, Order, Item, \ 1708 User, keywords, item_keywords = ( 1709 self.tables.users, 1710 self.classes.Keyword, 1711 self.tables.orders, 1712 self.tables.items, 1713 self.tables.order_items, 1714 self.classes.Order, 1715 self.classes.Item, 1716 self.classes.User, 1717 self.tables.keywords, 1718 self.tables.item_keywords) 1719 1720 mapper(User, users, properties={ 1721 'orders': relationship(Order, backref='user'), # o2m, m2o 1722 }) 1723 mapper(Order, orders, properties={ 1724 'items': relationship(Item, secondary=order_items, 1725 order_by=items.c.id), # m2m 1726 }) 1727 mapper(Item, items, properties={ 1728 'keywords': relationship(Keyword, secondary=item_keywords, 1729 order_by=keywords.c.id) # m2m 1730 }) 1731 mapper(Keyword, keywords) 1732 1733 sess = create_session() 1734 q = sess.query(User).join(User.orders).join(Order.items).\ 1735 options(joinedload_all("orders.items.keywords")) 1736 1737 # here, the eager join for keywords can catch onto 1738 # join(Order.items) or the nested (orders LEFT OUTER JOIN items), 1739 # it should catch the latter 1740 self.assert_compile( 1741 q, 1742 "SELECT users.id AS users_id, users.name AS users_name, " 1743 "keywords_1.id AS keywords_1_id, keywords_1.name " 1744 "AS keywords_1_name, " 1745 "items_1.id AS items_1_id, items_1.description AS " 1746 "items_1_description, " 1747 "orders_1.id AS orders_1_id, orders_1.user_id AS " 1748 "orders_1_user_id, " 1749 "orders_1.address_id AS orders_1_address_id, " 1750 "orders_1.description AS orders_1_description, " 1751 "orders_1.isopen AS orders_1_isopen " 1752 "FROM users JOIN orders ON users.id = orders.user_id " 1753 "JOIN order_items AS order_items_1 ON orders.id = " 1754 "order_items_1.order_id " 1755 "JOIN items ON items.id = order_items_1.item_id " 1756 "LEFT OUTER JOIN orders AS orders_1 ON users.id = " 1757 "orders_1.user_id " 1758 "LEFT OUTER JOIN (order_items AS order_items_2 " 1759 "JOIN items AS items_1 ON items_1.id = order_items_2.item_id) " 1760 "ON orders_1.id = order_items_2.order_id " 1761 "LEFT OUTER JOIN (item_keywords AS item_keywords_1 " 1762 "JOIN keywords AS keywords_1 ON keywords_1.id = " 1763 "item_keywords_1.keyword_id) " 1764 "ON items_1.id = item_keywords_1.item_id " 1765 "ORDER BY items_1.id, keywords_1.id" 1766 ) 1767 1768 def test_inner_join_unnested_chaining_fixed(self): 1769 users, items, order_items, Order, Item, User, orders = ( 1770 self.tables.users, 1771 self.tables.items, 1772 self.tables.order_items, 1773 self.classes.Order, 1774 self.classes.Item, 1775 self.classes.User, 1776 self.tables.orders) 1777 1778 mapper(User, users, properties=dict( 1779 orders=relationship(Order, lazy=False) 1780 )) 1781 mapper(Order, orders, properties=dict( 1782 items=relationship(Item, secondary=order_items, lazy=False, 1783 innerjoin="unnested") 1784 )) 1785 mapper(Item, items) 1786 1787 sess = create_session() 1788 1789 # joining from user, its all LEFT OUTER JOINs 1790 self.assert_compile( 1791 sess.query(User), 1792 "SELECT users.id AS users_id, users.name AS users_name, " 1793 "items_1.id AS " 1794 "items_1_id, items_1.description AS items_1_description, " 1795 "orders_1.id AS " 1796 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1797 "orders_1.address_id AS " 1798 "orders_1_address_id, orders_1.description AS " 1799 "orders_1_description, " 1800 "orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN " 1801 "orders AS orders_1 ON " 1802 "users.id = orders_1.user_id LEFT OUTER JOIN " 1803 "(order_items AS order_items_1 JOIN items AS items_1 ON " 1804 "items_1.id = " 1805 "order_items_1.item_id) ON orders_1.id = " 1806 "order_items_1.order_id" 1807 ) 1808 1809 # joining just from Order, innerjoin=True can be respected 1810 self.assert_compile( 1811 sess.query(Order), 1812 "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " 1813 "orders.address_id AS orders_address_id, orders.description AS " 1814 "orders_description, orders.isopen AS orders_isopen, items_1.id " 1815 "AS items_1_id, items_1.description AS items_1_description FROM " 1816 "orders JOIN order_items AS order_items_1 ON orders.id = " 1817 "order_items_1.order_id JOIN items AS items_1 ON items_1.id = " 1818 "order_items_1.item_id" 1819 ) 1820 1821 def test_inner_join_nested_chaining_fixed(self): 1822 users, items, order_items, Order, Item, User, orders = ( 1823 self.tables.users, 1824 self.tables.items, 1825 self.tables.order_items, 1826 self.classes.Order, 1827 self.classes.Item, 1828 self.classes.User, 1829 self.tables.orders) 1830 1831 mapper(User, users, properties=dict( 1832 orders=relationship(Order, lazy=False) 1833 )) 1834 mapper(Order, orders, properties=dict( 1835 items=relationship(Item, secondary=order_items, lazy=False, 1836 innerjoin='nested') 1837 )) 1838 mapper(Item, items) 1839 1840 sess = create_session() 1841 1842 self.assert_compile( 1843 sess.query(User), 1844 "SELECT users.id AS users_id, users.name AS users_name, " 1845 "items_1.id AS " 1846 "items_1_id, items_1.description AS items_1_description, " 1847 "orders_1.id AS " 1848 "orders_1_id, orders_1.user_id AS orders_1_user_id, " 1849 "orders_1.address_id AS " 1850 "orders_1_address_id, orders_1.description AS " 1851 "orders_1_description, " 1852 "orders_1.isopen AS orders_1_isopen " 1853 "FROM users LEFT OUTER JOIN " 1854 "(orders AS orders_1 JOIN order_items AS order_items_1 " 1855 "ON orders_1.id = order_items_1.order_id " 1856 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 1857 "ON users.id = orders_1.user_id" 1858 ) 1859 1860 def test_inner_join_options(self): 1861 users, items, order_items, Order, Item, User, orders = ( 1862 self.tables.users, 1863 self.tables.items, 1864 self.tables.order_items, 1865 self.classes.Order, 1866 self.classes.Item, 1867 self.classes.User, 1868 self.tables.orders) 1869 1870 mapper(User, users, properties=dict( 1871 orders=relationship(Order, backref=backref('user', innerjoin=True), 1872 order_by=orders.c.id) 1873 )) 1874 mapper(Order, orders, properties=dict( 1875 items=relationship( 1876 Item, 1877 secondary=order_items, 1878 order_by=items.c.id) 1879 )) 1880 mapper(Item, items) 1881 sess = create_session() 1882 self.assert_compile( 1883 sess.query(User).options(joinedload(User.orders, innerjoin=True)), 1884 "SELECT users.id AS users_id, users.name AS users_name, " 1885 "orders_1.id AS orders_1_id, " 1886 "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " 1887 "orders_1_address_id, " 1888 "orders_1.description AS orders_1_description, orders_1.isopen " 1889 "AS orders_1_isopen " 1890 "FROM users JOIN orders AS orders_1 ON users.id = " 1891 "orders_1.user_id ORDER BY orders_1.id") 1892 1893 self.assert_compile( 1894 sess.query(User).options( 1895 joinedload_all(User.orders, Order.items, innerjoin=True)), 1896 "SELECT users.id AS users_id, users.name AS users_name, " 1897 "items_1.id AS items_1_id, " 1898 "items_1.description AS items_1_description, " 1899 "orders_1.id AS orders_1_id, " 1900 "orders_1.user_id AS orders_1_user_id, orders_1.address_id " 1901 "AS orders_1_address_id, " 1902 "orders_1.description AS orders_1_description, orders_1.isopen " 1903 "AS orders_1_isopen " 1904 "FROM users JOIN orders AS orders_1 ON users.id = " 1905 "orders_1.user_id JOIN order_items AS " 1906 "order_items_1 ON orders_1.id = order_items_1.order_id " 1907 "JOIN items AS items_1 ON " 1908 "items_1.id = order_items_1.item_id ORDER BY orders_1.id, " 1909 "items_1.id") 1910 1911 def go(): 1912 eq_( 1913 sess.query(User).options( 1914 joinedload(User.orders, innerjoin=True), 1915 joinedload(User.orders, Order.items, innerjoin=True)). 1916 order_by(User.id).all(), 1917 1918 [User(id=7, 1919 orders=[ 1920 Order( 1921 id=1, items=[ 1922 Item( 1923 id=1), Item( 1924 id=2), Item( 1925 id=3)]), 1926 Order( 1927 id=3, items=[ 1928 Item( 1929 id=3), Item( 1930 id=4), Item( 1931 id=5)]), 1932 Order(id=5, items=[Item(id=5)])]), 1933 User(id=9, orders=[ 1934 Order( 1935 id=2, items=[ 1936 Item( 1937 id=1), Item( 1938 id=2), Item( 1939 id=3)]), 1940 Order(id=4, items=[Item(id=1), Item(id=5)])]) 1941 ] 1942 ) 1943 self.assert_sql_count(testing.db, go, 1) 1944 1945 # test that default innerjoin setting is used for options 1946 self.assert_compile( 1947 sess.query(Order).options( 1948 joinedload( 1949 Order.user)).filter( 1950 Order.description == 'foo'), 1951 "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " 1952 "orders.address_id AS " 1953 "orders_address_id, orders.description AS orders_description, " 1954 "orders.isopen AS " 1955 "orders_isopen, users_1.id AS users_1_id, users_1.name " 1956 "AS users_1_name " 1957 "FROM orders JOIN users AS users_1 ON users_1.id = orders.user_id " 1958 "WHERE orders.description = :description_1" 1959 ) 1960 1961 def test_propagated_lazyload_wildcard_unbound(self): 1962 self._test_propagated_lazyload_wildcard(False) 1963 1964 def test_propagated_lazyload_wildcard_bound(self): 1965 self._test_propagated_lazyload_wildcard(True) 1966 1967 def _test_propagated_lazyload_wildcard(self, use_load): 1968 users, items, order_items, Order, Item, User, orders = ( 1969 self.tables.users, 1970 self.tables.items, 1971 self.tables.order_items, 1972 self.classes.Order, 1973 self.classes.Item, 1974 self.classes.User, 1975 self.tables.orders) 1976 1977 mapper(User, users, properties=dict( 1978 orders=relationship(Order, lazy="select") 1979 )) 1980 mapper(Order, orders, properties=dict( 1981 items=relationship(Item, secondary=order_items, lazy="joined") 1982 )) 1983 mapper(Item, items) 1984 1985 sess = create_session() 1986 1987 if use_load: 1988 opt = Load(User).defaultload("orders").lazyload("*") 1989 else: 1990 opt = defaultload("orders").lazyload("*") 1991 1992 q = sess.query(User).filter(User.id == 7).options(opt) 1993 1994 def go(): 1995 for u in q: 1996 u.orders 1997 1998 self.sql_eq_(go, [ 1999 ("SELECT users.id AS users_id, users.name AS users_name " 2000 "FROM users WHERE users.id = :id_1", {"id_1": 7}), 2001 ("SELECT orders.id AS orders_id, " 2002 "orders.user_id AS orders_user_id, " 2003 "orders.address_id AS orders_address_id, " 2004 "orders.description AS orders_description, " 2005 "orders.isopen AS orders_isopen FROM orders " 2006 "WHERE :param_1 = orders.user_id", {"param_1": 7}), 2007 ]) 2008 2009 2010class InnerJoinSplicingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): 2011 __dialect__ = 'default' 2012 __backend__ = True # exercise hardcore join nesting on backends 2013 2014 @classmethod 2015 def define_tables(cls, metadata): 2016 Table('a', metadata, 2017 Column('id', Integer, primary_key=True) 2018 ) 2019 2020 Table('b', metadata, 2021 Column('id', Integer, primary_key=True), 2022 Column('a_id', Integer, ForeignKey('a.id')), 2023 Column('value', String(10)), 2024 ) 2025 Table('c1', metadata, 2026 Column('id', Integer, primary_key=True), 2027 Column('b_id', Integer, ForeignKey('b.id')), 2028 Column('value', String(10)), 2029 ) 2030 Table('c2', metadata, 2031 Column('id', Integer, primary_key=True), 2032 Column('b_id', Integer, ForeignKey('b.id')), 2033 Column('value', String(10)), 2034 ) 2035 Table('d1', metadata, 2036 Column('id', Integer, primary_key=True), 2037 Column('c1_id', Integer, ForeignKey('c1.id')), 2038 Column('value', String(10)), 2039 ) 2040 Table('d2', metadata, 2041 Column('id', Integer, primary_key=True), 2042 Column('c2_id', Integer, ForeignKey('c2.id')), 2043 Column('value', String(10)), 2044 ) 2045 Table('e1', metadata, 2046 Column('id', Integer, primary_key=True), 2047 Column('d1_id', Integer, ForeignKey('d1.id')), 2048 Column('value', String(10)), 2049 ) 2050 2051 @classmethod 2052 def setup_classes(cls): 2053 2054 class A(cls.Comparable): 2055 pass 2056 2057 class B(cls.Comparable): 2058 pass 2059 2060 class C1(cls.Comparable): 2061 pass 2062 2063 class C2(cls.Comparable): 2064 pass 2065 2066 class D1(cls.Comparable): 2067 pass 2068 2069 class D2(cls.Comparable): 2070 pass 2071 2072 class E1(cls.Comparable): 2073 pass 2074 2075 @classmethod 2076 def setup_mappers(cls): 2077 A, B, C1, C2, D1, D2, E1 = ( 2078 cls.classes.A, cls.classes.B, cls.classes.C1, 2079 cls.classes.C2, cls.classes.D1, cls.classes.D2, cls.classes.E1) 2080 mapper(A, cls.tables.a, properties={ 2081 'bs': relationship(B) 2082 }) 2083 mapper(B, cls.tables.b, properties=odict([ 2084 ('c1s', relationship(C1, order_by=cls.tables.c1.c.id)), 2085 ('c2s', relationship(C2, order_by=cls.tables.c2.c.id)) 2086 ])) 2087 mapper(C1, cls.tables.c1, properties={ 2088 'd1s': relationship(D1, order_by=cls.tables.d1.c.id) 2089 }) 2090 mapper(C2, cls.tables.c2, properties={ 2091 'd2s': relationship(D2, order_by=cls.tables.d2.c.id) 2092 }) 2093 mapper(D1, cls.tables.d1, properties={ 2094 'e1s': relationship(E1, order_by=cls.tables.e1.c.id) 2095 }) 2096 mapper(D2, cls.tables.d2) 2097 mapper(E1, cls.tables.e1) 2098 2099 @classmethod 2100 def _fixture_data(cls): 2101 A, B, C1, C2, D1, D2, E1 = ( 2102 cls.classes.A, cls.classes.B, cls.classes.C1, 2103 cls.classes.C2, cls.classes.D1, cls.classes.D2, cls.classes.E1) 2104 return [ 2105 A(id=1, bs=[ 2106 B( 2107 id=1, 2108 c1s=[C1( 2109 id=1, value='C11', 2110 d1s=[ 2111 D1(id=1, e1s=[E1(id=1)]), D1(id=2, e1s=[E1(id=2)]) 2112 ] 2113 ) 2114 ], 2115 c2s=[C2(id=1, value='C21', d2s=[D2(id=3)]), 2116 C2(id=2, value='C22', d2s=[D2(id=4)])] 2117 ), 2118 B( 2119 id=2, 2120 c1s=[ 2121 C1( 2122 id=4, value='C14', 2123 d1s=[D1( 2124 id=3, e1s=[ 2125 E1(id=3, value='E13'), 2126 E1(id=4, value="E14") 2127 ]), 2128 D1(id=4, e1s=[E1(id=5)]) 2129 ] 2130 ) 2131 ], 2132 c2s=[C2(id=4, value='C24', d2s=[])] 2133 ), 2134 ]), 2135 A(id=2, bs=[ 2136 B( 2137 id=3, 2138 c1s=[ 2139 C1( 2140 id=8, 2141 d1s=[D1(id=5, value='D15', e1s=[E1(id=6)])] 2142 ) 2143 ], 2144 c2s=[C2(id=8, d2s=[D2(id=6, value='D26')])] 2145 ) 2146 ]) 2147 ] 2148 2149 @classmethod 2150 def insert_data(cls): 2151 s = Session(testing.db) 2152 s.add_all(cls._fixture_data()) 2153 s.commit() 2154 2155 def _assert_result(self, query): 2156 eq_( 2157 query.all(), 2158 self._fixture_data() 2159 ) 2160 2161 def test_nested_innerjoin_propagation_multiple_paths_one(self): 2162 A, B, C1, C2 = ( 2163 self.classes.A, self.classes.B, self.classes.C1, 2164 self.classes.C2) 2165 2166 s = Session() 2167 2168 q = s.query(A).options( 2169 joinedload(A.bs, innerjoin=False). 2170 joinedload(B.c1s, innerjoin=True). 2171 joinedload(C1.d1s, innerjoin=True), 2172 defaultload(A.bs).joinedload(B.c2s, innerjoin=True). 2173 joinedload(C2.d2s, innerjoin=False) 2174 ) 2175 self.assert_compile( 2176 q, 2177 "SELECT a.id AS a_id, d1_1.id AS d1_1_id, " 2178 "d1_1.c1_id AS d1_1_c1_id, d1_1.value AS d1_1_value, " 2179 "c1_1.id AS c1_1_id, c1_1.b_id AS c1_1_b_id, " 2180 "c1_1.value AS c1_1_value, d2_1.id AS d2_1_id, " 2181 "d2_1.c2_id AS d2_1_c2_id, d2_1.value AS d2_1_value, " 2182 "c2_1.id AS c2_1_id, c2_1.b_id AS c2_1_b_id, " 2183 "c2_1.value AS c2_1_value, b_1.id AS b_1_id, " 2184 "b_1.a_id AS b_1_a_id, b_1.value AS b_1_value " 2185 "FROM a " 2186 "LEFT OUTER JOIN " 2187 "(b AS b_1 JOIN c2 AS c2_1 ON b_1.id = c2_1.b_id " 2188 "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id " 2189 "JOIN d1 AS d1_1 ON c1_1.id = d1_1.c1_id) ON a.id = b_1.a_id " 2190 "LEFT OUTER JOIN d2 AS d2_1 ON c2_1.id = d2_1.c2_id " 2191 "ORDER BY c1_1.id, d1_1.id, c2_1.id, d2_1.id" 2192 ) 2193 self._assert_result(q) 2194 2195 def test_nested_innerjoin_propagation_multiple_paths_two(self): 2196 # test #3447 2197 A = self.classes.A 2198 2199 s = Session() 2200 2201 q = s.query(A).options( 2202 joinedload('bs'), 2203 joinedload('bs.c2s', innerjoin=True), 2204 joinedload('bs.c1s', innerjoin=True), 2205 joinedload('bs.c1s.d1s') 2206 ) 2207 self.assert_compile( 2208 q, 2209 "SELECT a.id AS a_id, d1_1.id AS d1_1_id, " 2210 "d1_1.c1_id AS d1_1_c1_id, d1_1.value AS d1_1_value, " 2211 "c1_1.id AS c1_1_id, c1_1.b_id AS c1_1_b_id, " 2212 "c1_1.value AS c1_1_value, c2_1.id AS c2_1_id, " 2213 "c2_1.b_id AS c2_1_b_id, c2_1.value AS c2_1_value, " 2214 "b_1.id AS b_1_id, b_1.a_id AS b_1_a_id, " 2215 "b_1.value AS b_1_value " 2216 "FROM a LEFT OUTER JOIN " 2217 "(b AS b_1 JOIN c2 AS c2_1 ON b_1.id = c2_1.b_id " 2218 "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) ON a.id = b_1.a_id " 2219 "LEFT OUTER JOIN d1 AS d1_1 ON c1_1.id = d1_1.c1_id " 2220 "ORDER BY c1_1.id, d1_1.id, c2_1.id" 2221 ) 2222 self._assert_result(q) 2223 2224 def test_multiple_splice_points(self): 2225 A = self.classes.A 2226 2227 s = Session() 2228 2229 q = s.query(A).options( 2230 joinedload('bs', innerjoin=False), 2231 joinedload('bs.c1s', innerjoin=True), 2232 joinedload('bs.c2s', innerjoin=True), 2233 joinedload('bs.c1s.d1s', innerjoin=False), 2234 joinedload('bs.c2s.d2s'), 2235 joinedload('bs.c1s.d1s.e1s', innerjoin=True) 2236 ) 2237 2238 self.assert_compile( 2239 q, 2240 "SELECT a.id AS a_id, e1_1.id AS e1_1_id, " 2241 "e1_1.d1_id AS e1_1_d1_id, e1_1.value AS e1_1_value, " 2242 "d1_1.id AS d1_1_id, d1_1.c1_id AS d1_1_c1_id, " 2243 "d1_1.value AS d1_1_value, c1_1.id AS c1_1_id, " 2244 "c1_1.b_id AS c1_1_b_id, c1_1.value AS c1_1_value, " 2245 "d2_1.id AS d2_1_id, d2_1.c2_id AS d2_1_c2_id, " 2246 "d2_1.value AS d2_1_value, c2_1.id AS c2_1_id, " 2247 "c2_1.b_id AS c2_1_b_id, c2_1.value AS c2_1_value, " 2248 "b_1.id AS b_1_id, b_1.a_id AS b_1_a_id, b_1.value AS b_1_value " 2249 "FROM a LEFT OUTER JOIN " 2250 "(b AS b_1 JOIN c2 AS c2_1 ON b_1.id = c2_1.b_id " 2251 "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) ON a.id = b_1.a_id " 2252 "LEFT OUTER JOIN (" 2253 "d1 AS d1_1 JOIN e1 AS e1_1 ON d1_1.id = e1_1.d1_id) " 2254 "ON c1_1.id = d1_1.c1_id " 2255 "LEFT OUTER JOIN d2 AS d2_1 ON c2_1.id = d2_1.c2_id " 2256 "ORDER BY c1_1.id, d1_1.id, e1_1.id, c2_1.id, d2_1.id" 2257 ) 2258 self._assert_result(q) 2259 2260 def test_splice_onto_np_mapper(self): 2261 A = self.classes.A 2262 B = self.classes.B 2263 C1 = self.classes.C1 2264 b_table = self.tables.b 2265 c1_table = self.tables.c1 2266 2267 from sqlalchemy import inspect 2268 2269 weird_selectable = b_table.outerjoin(c1_table) 2270 2271 b_np = mapper( 2272 B, weird_selectable, non_primary=True, properties=odict([ 2273 # note we need to make this fixed with lazy=False until 2274 # [ticket:3348] is resolved 2275 ('c1s', relationship(C1, lazy=False, innerjoin=True)), 2276 ('c_id', c1_table.c.id), 2277 ('b_value', b_table.c.value), 2278 ]) 2279 ) 2280 2281 a_mapper = inspect(A) 2282 a_mapper.add_property( 2283 "bs_np", relationship(b_np) 2284 ) 2285 2286 s = Session() 2287 2288 q = s.query(A).options( 2289 joinedload('bs_np', innerjoin=False) 2290 ) 2291 self.assert_compile( 2292 q, 2293 "SELECT a.id AS a_id, c1_1.id AS c1_1_id, c1_1.b_id AS c1_1_b_id, " 2294 "c1_1.value AS c1_1_value, c1_2.id AS c1_2_id, " 2295 "b_1.value AS b_1_value, b_1.id AS b_1_id, " 2296 "b_1.a_id AS b_1_a_id, c1_2.b_id AS c1_2_b_id, " 2297 "c1_2.value AS c1_2_value " 2298 "FROM a LEFT OUTER JOIN " 2299 "(b AS b_1 LEFT OUTER JOIN c1 AS c1_2 ON b_1.id = c1_2.b_id " 2300 "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) ON a.id = b_1.a_id" 2301 ) 2302 2303 2304class InnerJoinSplicingWSecondaryTest( 2305 fixtures.MappedTest, testing.AssertsCompiledSQL): 2306 __dialect__ = 'default' 2307 __backend__ = True # exercise hardcore join nesting on backends 2308 2309 @classmethod 2310 def define_tables(cls, metadata): 2311 Table( 2312 'a', metadata, 2313 Column('id', Integer, primary_key=True), 2314 Column('bid', ForeignKey('b.id')) 2315 ) 2316 2317 Table( 2318 'b', metadata, 2319 Column('id', Integer, primary_key=True), 2320 Column('cid', ForeignKey('c.id')) 2321 ) 2322 2323 Table( 2324 'c', metadata, 2325 Column('id', Integer, primary_key=True), 2326 ) 2327 2328 Table('ctod', metadata, 2329 Column('cid', ForeignKey('c.id'), primary_key=True), 2330 Column('did', ForeignKey('d.id'), primary_key=True), 2331 ) 2332 Table('d', metadata, 2333 Column('id', Integer, primary_key=True), 2334 ) 2335 2336 @classmethod 2337 def setup_classes(cls): 2338 2339 class A(cls.Comparable): 2340 pass 2341 2342 class B(cls.Comparable): 2343 pass 2344 2345 class C(cls.Comparable): 2346 pass 2347 2348 class D(cls.Comparable): 2349 pass 2350 2351 @classmethod 2352 def setup_mappers(cls): 2353 A, B, C, D = ( 2354 cls.classes.A, cls.classes.B, cls.classes.C, 2355 cls.classes.D) 2356 mapper(A, cls.tables.a, properties={ 2357 'b': relationship(B) 2358 }) 2359 mapper(B, cls.tables.b, properties=odict([ 2360 ('c', relationship(C)), 2361 ])) 2362 mapper(C, cls.tables.c, properties=odict([ 2363 ('ds', relationship(D, secondary=cls.tables.ctod, 2364 order_by=cls.tables.d.c.id)), 2365 ])) 2366 mapper(D, cls.tables.d) 2367 2368 @classmethod 2369 def _fixture_data(cls): 2370 A, B, C, D = ( 2371 cls.classes.A, cls.classes.B, cls.classes.C, 2372 cls.classes.D) 2373 2374 d1, d2, d3 = D(id=1), D(id=2), D(id=3) 2375 return [ 2376 A( 2377 id=1, 2378 b=B( 2379 id=1, 2380 c=C( 2381 id=1, 2382 ds=[d1, d2] 2383 ) 2384 ) 2385 ), 2386 A( 2387 id=2, 2388 b=B( 2389 id=2, 2390 c=C( 2391 id=2, 2392 ds=[d2, d3] 2393 ) 2394 ) 2395 ) 2396 ] 2397 2398 @classmethod 2399 def insert_data(cls): 2400 s = Session(testing.db) 2401 s.add_all(cls._fixture_data()) 2402 s.commit() 2403 2404 def _assert_result(self, query): 2405 def go(): 2406 eq_( 2407 query.all(), 2408 self._fixture_data() 2409 ) 2410 2411 self.assert_sql_count( 2412 testing.db, 2413 go, 2414 1 2415 ) 2416 2417 def test_joined_across(self): 2418 A = self.classes.A 2419 2420 s = Session() 2421 q = s.query(A) \ 2422 .options( 2423 joinedload('b'). 2424 joinedload('c', innerjoin=True). 2425 joinedload('ds', innerjoin=True)) 2426 self.assert_compile( 2427 q, 2428 "SELECT a.id AS a_id, a.bid AS a_bid, d_1.id AS d_1_id, " 2429 "c_1.id AS c_1_id, b_1.id AS b_1_id, b_1.cid AS b_1_cid " 2430 "FROM a LEFT OUTER JOIN " 2431 "(b AS b_1 JOIN " 2432 "(c AS c_1 JOIN ctod AS ctod_1 ON c_1.id = ctod_1.cid) " 2433 "ON c_1.id = b_1.cid " 2434 "JOIN d AS d_1 ON d_1.id = ctod_1.did) ON b_1.id = a.bid " 2435 "ORDER BY d_1.id" 2436 ) 2437 self._assert_result(q) 2438 2439 2440class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): 2441 2442 """test #2188""" 2443 2444 __dialect__ = 'default' 2445 run_create_tables = None 2446 2447 @classmethod 2448 def define_tables(cls, metadata): 2449 Table('a', metadata, 2450 Column('id', Integer, primary_key=True) 2451 ) 2452 2453 Table('b', metadata, 2454 Column('id', Integer, primary_key=True), 2455 Column('a_id', Integer, ForeignKey('a.id')), 2456 Column('value', Integer), 2457 ) 2458 2459 @classmethod 2460 def setup_classes(cls): 2461 2462 class A(cls.Comparable): 2463 pass 2464 2465 class B(cls.Comparable): 2466 pass 2467 2468 def _fixture(self, props): 2469 A, B = self.classes.A, self.classes.B 2470 b_table, a_table = self.tables.b, self.tables.a 2471 mapper(A, a_table, properties=props) 2472 mapper(B, b_table, properties={ 2473 'a': relationship(A, backref="bs") 2474 }) 2475 2476 def test_column_property(self): 2477 A = self.classes.A 2478 b_table, a_table = self.tables.b, self.tables.a 2479 cp = select([func.sum(b_table.c.value)]).\ 2480 where(b_table.c.a_id == a_table.c.id) 2481 2482 self._fixture({ 2483 'summation': column_property(cp) 2484 }) 2485 self.assert_compile( 2486 create_session().query(A).options(joinedload_all('bs')). 2487 order_by(A.summation). 2488 limit(50), 2489 "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " 2490 "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " 2491 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " 2492 "(SELECT sum(b.value) AS sum_1 FROM b WHERE b.a_id = a.id) " 2493 "AS anon_2, a.id AS a_id FROM a ORDER BY anon_2 " 2494 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 ON " 2495 "anon_1.a_id = b_1.a_id ORDER BY anon_1.anon_2" 2496 ) 2497 2498 def test_column_property_desc(self): 2499 A = self.classes.A 2500 b_table, a_table = self.tables.b, self.tables.a 2501 cp = select([func.sum(b_table.c.value)]).\ 2502 where(b_table.c.a_id == a_table.c.id) 2503 2504 self._fixture({ 2505 'summation': column_property(cp) 2506 }) 2507 self.assert_compile( 2508 create_session().query(A).options(joinedload_all('bs')). 2509 order_by(A.summation.desc()). 2510 limit(50), 2511 "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " 2512 "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " 2513 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " 2514 "(SELECT sum(b.value) AS sum_1 FROM b WHERE b.a_id = a.id) " 2515 "AS anon_2, a.id AS a_id FROM a ORDER BY anon_2 DESC " 2516 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 ON " 2517 "anon_1.a_id = b_1.a_id ORDER BY anon_1.anon_2 DESC" 2518 ) 2519 2520 def test_column_property_correlated(self): 2521 A = self.classes.A 2522 b_table, a_table = self.tables.b, self.tables.a 2523 cp = select([func.sum(b_table.c.value)]).\ 2524 where(b_table.c.a_id == a_table.c.id).\ 2525 correlate(a_table) 2526 2527 self._fixture({ 2528 'summation': column_property(cp) 2529 }) 2530 self.assert_compile( 2531 create_session().query(A).options(joinedload_all('bs')). 2532 order_by(A.summation). 2533 limit(50), 2534 "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " 2535 "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " 2536 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " 2537 "(SELECT sum(b.value) AS sum_1 FROM b WHERE b.a_id = a.id) " 2538 "AS anon_2, a.id AS a_id FROM a ORDER BY anon_2 " 2539 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 ON " 2540 "anon_1.a_id = b_1.a_id ORDER BY anon_1.anon_2" 2541 ) 2542 2543 def test_standalone_subquery_unlabeled(self): 2544 A = self.classes.A 2545 b_table, a_table = self.tables.b, self.tables.a 2546 self._fixture({}) 2547 cp = select([func.sum(b_table.c.value)]).\ 2548 where(b_table.c.a_id == a_table.c.id).\ 2549 correlate(a_table).as_scalar() 2550 2551 # up until 0.8, this was ordering by a new subquery. 2552 # the removal of a separate _make_proxy() from ScalarSelect 2553 # fixed that. 2554 self.assert_compile( 2555 create_session().query(A).options(joinedload_all('bs')). 2556 order_by(cp). 2557 limit(50), 2558 "SELECT anon_1.a_id AS anon_1_a_id, anon_1.anon_2 " 2559 "AS anon_1_anon_2, b_1.id AS b_1_id, b_1.a_id AS " 2560 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " 2561 "AS a_id, (SELECT sum(b.value) AS sum_1 FROM b WHERE " 2562 "b.a_id = a.id) AS anon_2 FROM a ORDER BY (SELECT " 2563 "sum(b.value) AS sum_1 FROM b WHERE b.a_id = a.id) " 2564 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 " 2565 "ON anon_1.a_id = b_1.a_id ORDER BY anon_1.anon_2" 2566 ) 2567 2568 def test_standalone_subquery_labeled(self): 2569 A = self.classes.A 2570 b_table, a_table = self.tables.b, self.tables.a 2571 self._fixture({}) 2572 cp = select([func.sum(b_table.c.value)]).\ 2573 where(b_table.c.a_id == a_table.c.id).\ 2574 correlate(a_table).as_scalar().label('foo') 2575 self.assert_compile( 2576 create_session().query(A).options(joinedload_all('bs')). 2577 order_by(cp). 2578 limit(50), 2579 "SELECT anon_1.a_id AS anon_1_a_id, anon_1.foo " 2580 "AS anon_1_foo, b_1.id AS b_1_id, b_1.a_id AS " 2581 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " 2582 "AS a_id, (SELECT sum(b.value) AS sum_1 FROM b WHERE " 2583 "b.a_id = a.id) AS foo FROM a ORDER BY foo " 2584 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 " 2585 "ON anon_1.a_id = b_1.a_id ORDER BY " 2586 "anon_1.foo" 2587 ) 2588 2589 def test_standalone_negated(self): 2590 A = self.classes.A 2591 b_table, a_table = self.tables.b, self.tables.a 2592 self._fixture({}) 2593 cp = select([func.sum(b_table.c.value)]).\ 2594 where(b_table.c.a_id == a_table.c.id).\ 2595 correlate(a_table).\ 2596 as_scalar() 2597 # test a different unary operator 2598 self.assert_compile( 2599 create_session().query(A).options(joinedload_all('bs')). 2600 order_by(~cp). 2601 limit(50), 2602 "SELECT anon_1.a_id AS anon_1_a_id, anon_1.anon_2 " 2603 "AS anon_1_anon_2, b_1.id AS b_1_id, b_1.a_id AS " 2604 "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " 2605 "AS a_id, NOT (SELECT sum(b.value) AS sum_1 FROM b " 2606 "WHERE b.a_id = a.id) FROM a ORDER BY NOT (SELECT " 2607 "sum(b.value) AS sum_1 FROM b WHERE b.a_id = a.id) " 2608 "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN b AS b_1 " 2609 "ON anon_1.a_id = b_1.a_id ORDER BY anon_1.anon_2" 2610 ) 2611 2612 2613class LoadOnExistingTest(_fixtures.FixtureTest): 2614 2615 """test that loaders from a base Query fully populate.""" 2616 2617 run_inserts = 'once' 2618 run_deletes = None 2619 2620 def _collection_to_scalar_fixture(self): 2621 User, Address, Dingaling = self.classes.User, \ 2622 self.classes.Address, self.classes.Dingaling 2623 mapper(User, self.tables.users, properties={ 2624 'addresses': relationship(Address), 2625 }) 2626 mapper(Address, self.tables.addresses, properties={ 2627 'dingaling': relationship(Dingaling) 2628 }) 2629 mapper(Dingaling, self.tables.dingalings) 2630 2631 sess = Session(autoflush=False) 2632 return User, Address, Dingaling, sess 2633 2634 def _collection_to_collection_fixture(self): 2635 User, Order, Item = self.classes.User, \ 2636 self.classes.Order, self.classes.Item 2637 mapper(User, self.tables.users, properties={ 2638 'orders': relationship(Order), 2639 }) 2640 mapper(Order, self.tables.orders, properties={ 2641 'items': relationship(Item, secondary=self.tables.order_items), 2642 }) 2643 mapper(Item, self.tables.items) 2644 2645 sess = Session(autoflush=False) 2646 return User, Order, Item, sess 2647 2648 def _eager_config_fixture(self): 2649 User, Address = self.classes.User, self.classes.Address 2650 mapper(User, self.tables.users, properties={ 2651 'addresses': relationship(Address, lazy="joined"), 2652 }) 2653 mapper(Address, self.tables.addresses) 2654 sess = Session(autoflush=False) 2655 return User, Address, sess 2656 2657 def test_no_query_on_refresh(self): 2658 User, Address, sess = self._eager_config_fixture() 2659 2660 u1 = sess.query(User).get(8) 2661 assert 'addresses' in u1.__dict__ 2662 sess.expire(u1) 2663 2664 def go(): 2665 eq_(u1.id, 8) 2666 self.assert_sql_count(testing.db, go, 1) 2667 assert 'addresses' not in u1.__dict__ 2668 2669 def test_loads_second_level_collection_to_scalar(self): 2670 User, Address, Dingaling, sess = self._collection_to_scalar_fixture() 2671 2672 u1 = sess.query(User).get(8) 2673 a1 = Address() 2674 u1.addresses.append(a1) 2675 a2 = u1.addresses[0] 2676 a2.email_address = 'foo' 2677 sess.query(User).options(joinedload_all("addresses.dingaling")).\ 2678 filter_by(id=8).all() 2679 assert u1.addresses[-1] is a1 2680 for a in u1.addresses: 2681 if a is not a1: 2682 assert 'dingaling' in a.__dict__ 2683 else: 2684 assert 'dingaling' not in a.__dict__ 2685 if a is a2: 2686 eq_(a2.email_address, 'foo') 2687 2688 def test_loads_second_level_collection_to_collection(self): 2689 User, Order, Item, sess = self._collection_to_collection_fixture() 2690 2691 u1 = sess.query(User).get(7) 2692 u1.orders 2693 o1 = Order() 2694 u1.orders.append(o1) 2695 sess.query(User).options(joinedload_all("orders.items")).\ 2696 filter_by(id=7).all() 2697 for o in u1.orders: 2698 if o is not o1: 2699 assert 'items' in o.__dict__ 2700 else: 2701 assert 'items' not in o.__dict__ 2702 2703 def test_load_two_levels_collection_to_scalar(self): 2704 User, Address, Dingaling, sess = self._collection_to_scalar_fixture() 2705 2706 u1 = sess.query(User).filter_by( 2707 id=8).options( 2708 joinedload("addresses")).one() 2709 sess.query(User).filter_by( 2710 id=8).options( 2711 joinedload_all("addresses.dingaling")).first() 2712 assert 'dingaling' in u1.addresses[0].__dict__ 2713 2714 def test_load_two_levels_collection_to_collection(self): 2715 User, Order, Item, sess = self._collection_to_collection_fixture() 2716 2717 u1 = sess.query(User).filter_by( 2718 id=7).options( 2719 joinedload("orders")).one() 2720 sess.query(User).filter_by( 2721 id=7).options( 2722 joinedload_all("orders.items")).first() 2723 assert 'items' in u1.orders[0].__dict__ 2724 2725 2726class AddEntityTest(_fixtures.FixtureTest): 2727 run_inserts = 'once' 2728 run_deletes = None 2729 2730 def _assert_result(self): 2731 Item, Address, Order, User = (self.classes.Item, 2732 self.classes.Address, 2733 self.classes.Order, 2734 self.classes.User) 2735 2736 return [ 2737 ( 2738 User(id=7, 2739 addresses=[Address(id=1)] 2740 ), 2741 Order(id=1, 2742 items=[Item(id=1), Item(id=2), Item(id=3)] 2743 ), 2744 ), 2745 ( 2746 User(id=7, 2747 addresses=[Address(id=1)] 2748 ), 2749 Order(id=3, 2750 items=[Item(id=3), Item(id=4), Item(id=5)] 2751 ), 2752 ), 2753 ( 2754 User(id=7, 2755 addresses=[Address(id=1)] 2756 ), 2757 Order(id=5, 2758 items=[Item(id=5)] 2759 ), 2760 ), 2761 ( 2762 User(id=9, 2763 addresses=[Address(id=5)] 2764 ), 2765 Order(id=2, 2766 items=[Item(id=1), Item(id=2), Item(id=3)] 2767 ), 2768 ), 2769 ( 2770 User(id=9, 2771 addresses=[Address(id=5)] 2772 ), 2773 Order(id=4, 2774 items=[Item(id=1), Item(id=5)] 2775 ), 2776 ) 2777 ] 2778 2779 def test_mapper_configured(self): 2780 users, items, order_items, Order, \ 2781 Item, User, Address, orders, addresses = ( 2782 self.tables.users, 2783 self.tables.items, 2784 self.tables.order_items, 2785 self.classes.Order, 2786 self.classes.Item, 2787 self.classes.User, 2788 self.classes.Address, 2789 self.tables.orders, 2790 self.tables.addresses) 2791 2792 mapper(User, users, properties={ 2793 'addresses': relationship(Address, lazy='joined'), 2794 'orders': relationship(Order) 2795 }) 2796 mapper(Address, addresses) 2797 mapper(Order, orders, properties={ 2798 'items': relationship( 2799 Item, secondary=order_items, lazy='joined', 2800 order_by=items.c.id) 2801 }) 2802 mapper(Item, items) 2803 2804 sess = create_session() 2805 oalias = sa.orm.aliased(Order) 2806 2807 def go(): 2808 ret = sess.query(User, oalias).join(oalias, 'orders').\ 2809 order_by(User.id, oalias.id).all() 2810 eq_(ret, self._assert_result()) 2811 self.assert_sql_count(testing.db, go, 1) 2812 2813 def test_options(self): 2814 users, items, order_items, Order,\ 2815 Item, User, Address, orders, addresses = ( 2816 self.tables.users, 2817 self.tables.items, 2818 self.tables.order_items, 2819 self.classes.Order, 2820 self.classes.Item, 2821 self.classes.User, 2822 self.classes.Address, 2823 self.tables.orders, 2824 self.tables.addresses) 2825 2826 mapper(User, users, properties={ 2827 'addresses': relationship(Address), 2828 'orders': relationship(Order) 2829 }) 2830 mapper(Address, addresses) 2831 mapper(Order, orders, properties={ 2832 'items': relationship( 2833 Item, secondary=order_items, order_by=items.c.id) 2834 }) 2835 mapper(Item, items) 2836 2837 sess = create_session() 2838 2839 oalias = sa.orm.aliased(Order) 2840 2841 def go(): 2842 ret = sess.query(User, oalias).options(joinedload('addresses')).\ 2843 join(oalias, 'orders').\ 2844 order_by(User.id, oalias.id).all() 2845 eq_(ret, self._assert_result()) 2846 self.assert_sql_count(testing.db, go, 6) 2847 2848 sess.expunge_all() 2849 2850 def go(): 2851 ret = sess.query(User, oalias).\ 2852 options(joinedload('addresses'), 2853 joinedload(oalias.items)).\ 2854 join(oalias, 'orders').\ 2855 order_by(User.id, oalias.id).all() 2856 eq_(ret, self._assert_result()) 2857 self.assert_sql_count(testing.db, go, 1) 2858 2859 2860class OrderBySecondaryTest(fixtures.MappedTest): 2861 2862 @classmethod 2863 def define_tables(cls, metadata): 2864 Table('m2m', metadata, 2865 Column( 2866 'id', 2867 Integer, 2868 primary_key=True, 2869 test_needs_autoincrement=True), 2870 Column('aid', Integer, ForeignKey('a.id')), 2871 Column('bid', Integer, ForeignKey('b.id'))) 2872 2873 Table('a', metadata, 2874 Column( 2875 'id', 2876 Integer, 2877 primary_key=True, 2878 test_needs_autoincrement=True), 2879 Column('data', String(50))) 2880 Table('b', metadata, 2881 Column( 2882 'id', 2883 Integer, 2884 primary_key=True, 2885 test_needs_autoincrement=True), 2886 Column('data', String(50))) 2887 2888 @classmethod 2889 def fixtures(cls): 2890 return dict( 2891 a=(('id', 'data'), 2892 (1, 'a1'), 2893 (2, 'a2')), 2894 2895 b=(('id', 'data'), 2896 (1, 'b1'), 2897 (2, 'b2'), 2898 (3, 'b3'), 2899 (4, 'b4')), 2900 2901 m2m=(('id', 'aid', 'bid'), 2902 (2, 1, 1), 2903 (4, 2, 4), 2904 (1, 1, 3), 2905 (6, 2, 2), 2906 (3, 1, 2), 2907 (5, 2, 3))) 2908 2909 def test_ordering(self): 2910 a, m2m, b = ( 2911 self.tables.a, 2912 self.tables.m2m, 2913 self.tables.b) 2914 2915 class A(fixtures.ComparableEntity): 2916 pass 2917 2918 class B(fixtures.ComparableEntity): 2919 pass 2920 2921 mapper(A, a, properties={ 2922 'bs': relationship( 2923 B, secondary=m2m, lazy='joined', order_by=m2m.c.id) 2924 }) 2925 mapper(B, b) 2926 2927 sess = create_session() 2928 eq_(sess.query(A).all(), 2929 [ 2930 A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), 2931 A(bs=[B(data='b4'), B(data='b3'), B(data='b2')]) 2932 ]) 2933 2934 2935class SelfReferentialEagerTest(fixtures.MappedTest): 2936 2937 @classmethod 2938 def define_tables(cls, metadata): 2939 Table('nodes', metadata, 2940 Column( 2941 'id', 2942 Integer, 2943 primary_key=True, 2944 test_needs_autoincrement=True), 2945 Column('parent_id', Integer, ForeignKey('nodes.id')), 2946 Column('data', String(30))) 2947 2948 def test_basic(self): 2949 nodes = self.tables.nodes 2950 2951 class Node(fixtures.ComparableEntity): 2952 2953 def append(self, node): 2954 self.children.append(node) 2955 2956 mapper(Node, nodes, properties={ 2957 'children': relationship(Node, 2958 lazy='joined', 2959 join_depth=3, order_by=nodes.c.id) 2960 }) 2961 sess = create_session() 2962 n1 = Node(data='n1') 2963 n1.append(Node(data='n11')) 2964 n1.append(Node(data='n12')) 2965 n1.append(Node(data='n13')) 2966 n1.children[1].append(Node(data='n121')) 2967 n1.children[1].append(Node(data='n122')) 2968 n1.children[1].append(Node(data='n123')) 2969 sess.add(n1) 2970 sess.flush() 2971 sess.expunge_all() 2972 2973 def go(): 2974 d = sess.query(Node).filter_by(data='n1').all()[0] 2975 eq_(Node(data='n1', children=[ 2976 Node(data='n11'), 2977 Node(data='n12', children=[ 2978 Node(data='n121'), 2979 Node(data='n122'), 2980 Node(data='n123') 2981 ]), 2982 Node(data='n13') 2983 ]), d) 2984 self.assert_sql_count(testing.db, go, 1) 2985 2986 sess.expunge_all() 2987 2988 def go(): 2989 d = sess.query(Node).filter_by(data='n1').first() 2990 eq_(Node(data='n1', children=[ 2991 Node(data='n11'), 2992 Node(data='n12', children=[ 2993 Node(data='n121'), 2994 Node(data='n122'), 2995 Node(data='n123') 2996 ]), 2997 Node(data='n13') 2998 ]), d) 2999 self.assert_sql_count(testing.db, go, 1) 3000 3001 def test_lazy_fallback_doesnt_affect_eager(self): 3002 nodes = self.tables.nodes 3003 3004 class Node(fixtures.ComparableEntity): 3005 3006 def append(self, node): 3007 self.children.append(node) 3008 3009 mapper(Node, nodes, properties={ 3010 'children': relationship(Node, lazy='joined', join_depth=1, 3011 order_by=nodes.c.id) 3012 }) 3013 sess = create_session() 3014 n1 = Node(data='n1') 3015 n1.append(Node(data='n11')) 3016 n1.append(Node(data='n12')) 3017 n1.append(Node(data='n13')) 3018 n1.children[1].append(Node(data='n121')) 3019 n1.children[1].append(Node(data='n122')) 3020 n1.children[1].append(Node(data='n123')) 3021 sess.add(n1) 3022 sess.flush() 3023 sess.expunge_all() 3024 3025 # eager load with join depth 1. when eager load of 'n1' hits the 3026 # children of 'n12', no columns are present, eager loader degrades to 3027 # lazy loader; fine. but then, 'n12' is *also* in the first level of 3028 # columns since we're loading the whole table. when those rows 3029 # arrive, now we *can* eager load its children and an eager collection 3030 # should be initialized. essentially the 'n12' instance is present in 3031 # not just two different rows but two distinct sets of columns in this 3032 # result set. 3033 def go(): 3034 allnodes = sess.query(Node).order_by(Node.data).all() 3035 n12 = allnodes[2] 3036 eq_(n12.data, 'n12') 3037 eq_([ 3038 Node(data='n121'), 3039 Node(data='n122'), 3040 Node(data='n123') 3041 ], list(n12.children)) 3042 self.assert_sql_count(testing.db, go, 1) 3043 3044 def test_with_deferred(self): 3045 nodes = self.tables.nodes 3046 3047 class Node(fixtures.ComparableEntity): 3048 3049 def append(self, node): 3050 self.children.append(node) 3051 3052 mapper(Node, nodes, properties={ 3053 'children': relationship(Node, lazy='joined', join_depth=3, 3054 order_by=nodes.c.id), 3055 'data': deferred(nodes.c.data) 3056 }) 3057 sess = create_session() 3058 n1 = Node(data='n1') 3059 n1.append(Node(data='n11')) 3060 n1.append(Node(data='n12')) 3061 sess.add(n1) 3062 sess.flush() 3063 sess.expunge_all() 3064 3065 def go(): 3066 eq_( 3067 Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), 3068 sess.query(Node).order_by(Node.id).first(), 3069 ) 3070 self.assert_sql_count(testing.db, go, 4) 3071 3072 sess.expunge_all() 3073 3074 def go(): 3075 eq_(Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), 3076 sess.query(Node). 3077 options(undefer('data')).order_by(Node.id).first()) 3078 self.assert_sql_count(testing.db, go, 3) 3079 3080 sess.expunge_all() 3081 3082 def go(): 3083 eq_(Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), 3084 sess.query(Node).options(undefer('data'), 3085 undefer('children.data')).first()) 3086 self.assert_sql_count(testing.db, go, 1) 3087 3088 def test_options(self): 3089 nodes = self.tables.nodes 3090 3091 class Node(fixtures.ComparableEntity): 3092 3093 def append(self, node): 3094 self.children.append(node) 3095 3096 mapper(Node, nodes, properties={ 3097 'children': relationship(Node, lazy='select', order_by=nodes.c.id) 3098 }, order_by=nodes.c.id) 3099 sess = create_session() 3100 n1 = Node(data='n1') 3101 n1.append(Node(data='n11')) 3102 n1.append(Node(data='n12')) 3103 n1.append(Node(data='n13')) 3104 n1.children[1].append(Node(data='n121')) 3105 n1.children[1].append(Node(data='n122')) 3106 n1.children[1].append(Node(data='n123')) 3107 sess.add(n1) 3108 sess.flush() 3109 sess.expunge_all() 3110 3111 def go(): 3112 d = sess.query(Node).filter_by(data='n1').\ 3113 options(joinedload('children.children')).first() 3114 eq_(Node(data='n1', children=[ 3115 Node(data='n11'), 3116 Node(data='n12', children=[ 3117 Node(data='n121'), 3118 Node(data='n122'), 3119 Node(data='n123') 3120 ]), 3121 Node(data='n13') 3122 ]), d) 3123 self.assert_sql_count(testing.db, go, 2) 3124 3125 def go(): 3126 sess.query(Node).filter_by(data='n1').\ 3127 options(joinedload('children.children')).first() 3128 3129 # test that the query isn't wrapping the initial query for eager 3130 # loading. 3131 self.assert_sql_execution( 3132 testing.db, go, 3133 CompiledSQL( 3134 "SELECT nodes.id AS nodes_id, nodes.parent_id AS " 3135 "nodes_parent_id, nodes.data AS nodes_data FROM nodes " 3136 "WHERE nodes.data = :data_1 ORDER BY nodes.id LIMIT :param_1", 3137 {'data_1': 'n1'} 3138 ) 3139 ) 3140 3141 def test_no_depth(self): 3142 nodes = self.tables.nodes 3143 3144 class Node(fixtures.ComparableEntity): 3145 3146 def append(self, node): 3147 self.children.append(node) 3148 3149 mapper(Node, nodes, properties={ 3150 'children': relationship(Node, lazy='joined') 3151 }) 3152 sess = create_session() 3153 n1 = Node(data='n1') 3154 n1.append(Node(data='n11')) 3155 n1.append(Node(data='n12')) 3156 n1.append(Node(data='n13')) 3157 n1.children[1].append(Node(data='n121')) 3158 n1.children[1].append(Node(data='n122')) 3159 n1.children[1].append(Node(data='n123')) 3160 sess.add(n1) 3161 sess.flush() 3162 sess.expunge_all() 3163 3164 def go(): 3165 d = sess.query(Node).filter_by(data='n1').first() 3166 eq_(Node(data='n1', children=[ 3167 Node(data='n11'), 3168 Node(data='n12', children=[ 3169 Node(data='n121'), 3170 Node(data='n122'), 3171 Node(data='n123') 3172 ]), 3173 Node(data='n13') 3174 ]), d) 3175 self.assert_sql_count(testing.db, go, 3) 3176 3177 3178class MixedSelfReferentialEagerTest(fixtures.MappedTest): 3179 3180 @classmethod 3181 def define_tables(cls, metadata): 3182 Table('a_table', metadata, 3183 Column( 3184 'id', 3185 Integer, 3186 primary_key=True, 3187 test_needs_autoincrement=True) 3188 ) 3189 3190 Table('b_table', metadata, 3191 Column( 3192 'id', 3193 Integer, 3194 primary_key=True, 3195 test_needs_autoincrement=True), 3196 Column('parent_b1_id', Integer, ForeignKey('b_table.id')), 3197 Column('parent_a_id', Integer, ForeignKey('a_table.id')), 3198 Column('parent_b2_id', Integer, ForeignKey('b_table.id'))) 3199 3200 @classmethod 3201 def setup_mappers(cls): 3202 b_table, a_table = cls.tables.b_table, cls.tables.a_table 3203 3204 class A(cls.Comparable): 3205 pass 3206 3207 class B(cls.Comparable): 3208 pass 3209 3210 mapper(A, a_table) 3211 mapper(B, b_table, properties={ 3212 'parent_b1': relationship( 3213 B, 3214 remote_side=[b_table.c.id], 3215 primaryjoin=(b_table.c.parent_b1_id == b_table.c.id), 3216 order_by=b_table.c.id 3217 ), 3218 'parent_z': relationship(A, lazy=True), 3219 'parent_b2': relationship( 3220 B, 3221 remote_side=[b_table.c.id], 3222 primaryjoin=(b_table.c.parent_b2_id == b_table.c.id), 3223 order_by = b_table.c.id 3224 ) 3225 }) 3226 3227 @classmethod 3228 def insert_data(cls): 3229 b_table, a_table = cls.tables.b_table, cls.tables.a_table 3230 3231 a_table.insert().execute(dict(id=1), dict(id=2), dict(id=3)) 3232 b_table.insert().execute( 3233 dict(id=1, parent_a_id=2, parent_b1_id=None, parent_b2_id=None), 3234 dict(id=2, parent_a_id=1, parent_b1_id=1, parent_b2_id=None), 3235 dict(id=3, parent_a_id=1, parent_b1_id=1, parent_b2_id=2), 3236 dict(id=4, parent_a_id=3, parent_b1_id=1, parent_b2_id=None), 3237 dict(id=5, parent_a_id=3, parent_b1_id=None, parent_b2_id=2), 3238 dict(id=6, parent_a_id=1, parent_b1_id=1, parent_b2_id=3), 3239 dict(id=7, parent_a_id=2, parent_b1_id=None, parent_b2_id=3), 3240 dict(id=8, parent_a_id=2, parent_b1_id=1, parent_b2_id=2), 3241 dict(id=9, parent_a_id=None, parent_b1_id=1, parent_b2_id=None), 3242 dict(id=10, parent_a_id=3, parent_b1_id=7, parent_b2_id=2), 3243 dict(id=11, parent_a_id=3, parent_b1_id=1, parent_b2_id=8), 3244 dict(id=12, parent_a_id=2, parent_b1_id=5, parent_b2_id=2), 3245 dict(id=13, parent_a_id=3, parent_b1_id=4, parent_b2_id=4), 3246 dict(id=14, parent_a_id=3, parent_b1_id=7, parent_b2_id=2), 3247 ) 3248 3249 def test_eager_load(self): 3250 A, B = self.classes.A, self.classes.B 3251 3252 session = create_session() 3253 3254 def go(): 3255 eq_( 3256 session.query(B). 3257 options( 3258 joinedload('parent_b1'), 3259 joinedload('parent_b2'), 3260 joinedload('parent_z') 3261 ). 3262 filter(B.id.in_([2, 8, 11])).order_by(B.id).all(), 3263 [ 3264 B(id=2, 3265 parent_z=A(id=1), 3266 parent_b1=B(id=1), 3267 parent_b2=None), 3268 B(id=8, 3269 parent_z=A(id=2), 3270 parent_b1=B(id=1), 3271 parent_b2=B(id=2)), 3272 B(id=11, 3273 parent_z=A(id=3), 3274 parent_b1=B(id=1), 3275 parent_b2=B(id=8)) 3276 ] 3277 ) 3278 self.assert_sql_count(testing.db, go, 1) 3279 3280 3281class SelfReferentialM2MEagerTest(fixtures.MappedTest): 3282 3283 @classmethod 3284 def define_tables(cls, metadata): 3285 Table('widget', metadata, 3286 Column( 3287 'id', 3288 Integer, 3289 primary_key=True, 3290 test_needs_autoincrement=True), 3291 Column('name', sa.String(40), nullable=False, unique=True), 3292 ) 3293 3294 Table('widget_rel', metadata, 3295 Column('parent_id', Integer, ForeignKey('widget.id')), 3296 Column('child_id', Integer, ForeignKey('widget.id')), 3297 sa.UniqueConstraint('parent_id', 'child_id'), 3298 ) 3299 3300 def test_basic(self): 3301 widget, widget_rel = self.tables.widget, self.tables.widget_rel 3302 3303 class Widget(fixtures.ComparableEntity): 3304 pass 3305 3306 mapper(Widget, widget, properties={ 3307 'children': relationship( 3308 Widget, secondary=widget_rel, 3309 primaryjoin=widget_rel.c.parent_id == widget.c.id, 3310 secondaryjoin=widget_rel.c.child_id == widget.c.id, 3311 lazy='joined', join_depth=1, 3312 ) 3313 }) 3314 3315 sess = create_session() 3316 w1 = Widget(name='w1') 3317 w2 = Widget(name='w2') 3318 w1.children.append(w2) 3319 sess.add(w1) 3320 sess.flush() 3321 sess.expunge_all() 3322 3323 eq_([Widget(name='w1', children=[Widget(name='w2')])], 3324 sess.query(Widget).filter(Widget.name == 'w1').all()) 3325 3326 3327class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): 3328 run_setup_mappers = 'once' 3329 run_inserts = 'once' 3330 run_deletes = None 3331 __dialect__ = 'default' 3332 3333 __prefer_backends__ = ('postgresql', 'mysql', 'oracle') 3334 3335 @classmethod 3336 def setup_mappers(cls): 3337 users, Keyword, items, order_items, orders, \ 3338 Item, User, Address, keywords, Order, \ 3339 item_keywords, addresses = ( 3340 cls.tables.users, 3341 cls.classes.Keyword, 3342 cls.tables.items, 3343 cls.tables.order_items, 3344 cls.tables.orders, 3345 cls.classes.Item, 3346 cls.classes.User, 3347 cls.classes.Address, 3348 cls.tables.keywords, 3349 cls.classes.Order, 3350 cls.tables.item_keywords, 3351 cls.tables.addresses) 3352 3353 mapper(User, users, properties={ 3354 'addresses': relationship(Address, backref='user'), 3355 'orders': relationship(Order, backref='user'), # o2m, m2o 3356 }) 3357 mapper(Address, addresses) 3358 mapper(Order, orders, properties={ 3359 'items': relationship( 3360 Item, secondary=order_items, order_by=items.c.id), # m2m 3361 }) 3362 mapper(Item, items, properties={ 3363 'keywords': relationship(Keyword, secondary=item_keywords) # m2m 3364 }) 3365 mapper(Keyword, keywords) 3366 3367 def test_two_entities(self): 3368 Item, Order, User, Address = (self.classes.Item, 3369 self.classes.Order, 3370 self.classes.User, 3371 self.classes.Address) 3372 3373 sess = create_session() 3374 3375 # two FROM clauses 3376 def go(): 3377 eq_( 3378 [ 3379 (User(id=9, addresses=[Address(id=5)]), 3380 Order(id=2, items=[ 3381 Item(id=1), Item(id=2), Item(id=3)])), 3382 (User(id=9, addresses=[Address(id=5)]), 3383 Order(id=4, items=[ 3384 Item(id=1), Item(id=5)])), 3385 ], 3386 sess.query(User, Order).filter(User.id == Order.user_id). 3387 options(joinedload(User.addresses), joinedload(Order.items)). 3388 filter(User.id == 9). 3389 order_by(User.id, Order.id).all(), 3390 ) 3391 self.assert_sql_count(testing.db, go, 1) 3392 3393 # one FROM clause 3394 def go(): 3395 eq_( 3396 [ 3397 (User(id=9, addresses=[Address(id=5)]), 3398 Order(id=2, items=[ 3399 Item(id=1), Item(id=2), Item(id=3)])), 3400 (User(id=9, addresses=[Address(id=5)]), 3401 Order(id=4, items=[ 3402 Item(id=1), Item(id=5)])), 3403 ], 3404 sess.query(User, Order).join(User.orders). 3405 options(joinedload(User.addresses), joinedload(Order.items)). 3406 filter(User.id == 9). 3407 order_by(User.id, Order.id).all(), 3408 ) 3409 self.assert_sql_count(testing.db, go, 1) 3410 3411 @testing.exclude( 3412 'sqlite', '>', (0, ), "sqlite flat out blows it on the multiple JOINs") 3413 def test_two_entities_with_joins(self): 3414 Item, Order, User, Address = (self.classes.Item, 3415 self.classes.Order, 3416 self.classes.User, 3417 self.classes.Address) 3418 3419 sess = create_session() 3420 3421 # two FROM clauses where there's a join on each one 3422 def go(): 3423 u1 = aliased(User) 3424 o1 = aliased(Order) 3425 eq_( 3426 [ 3427 ( 3428 User(addresses=[ 3429 Address(email_address='fred@fred.com')], 3430 name='fred'), 3431 Order(description='order 2', isopen=0, 3432 items=[ 3433 Item(description='item 1'), 3434 Item(description='item 2'), 3435 Item(description='item 3')]), 3436 User(addresses=[ 3437 Address(email_address='jack@bean.com')], 3438 name='jack'), 3439 Order(description='order 3', isopen=1, 3440 items=[ 3441 Item(description='item 3'), 3442 Item(description='item 4'), 3443 Item(description='item 5')]) 3444 ), 3445 3446 ( 3447 User( 3448 addresses=[ 3449 Address( 3450 email_address='fred@fred.com')], 3451 name='fred'), 3452 Order( 3453 description='order 2', isopen=0, items=[ 3454 Item( 3455 description='item 1'), Item( 3456 description='item 2'), Item( 3457 description='item 3')]), 3458 User( 3459 addresses=[ 3460 Address( 3461 email_address='jack@bean.com')], 3462 name='jack'), 3463 Order( 3464 address_id=None, 3465 description='order 5', 3466 isopen=0, 3467 items=[ 3468 Item( 3469 description='item 5')]) 3470 ), 3471 3472 ( 3473 User( 3474 addresses=[ 3475 Address( 3476 email_address='fred@fred.com')], 3477 name='fred'), 3478 Order( 3479 description='order 4', isopen=1, items=[ 3480 Item( 3481 description='item 1'), Item( 3482 description='item 5')]), 3483 User( 3484 addresses=[ 3485 Address( 3486 email_address='jack@bean.com')], 3487 name='jack'), 3488 Order( 3489 address_id=None, 3490 description='order 5', 3491 isopen=0, 3492 items=[ 3493 Item( 3494 description='item 5')]) 3495 ), 3496 ], 3497 sess.query(User, Order, u1, o1). 3498 join(Order, User.orders). 3499 options(joinedload(User.addresses), 3500 joinedload(Order.items)).filter(User.id == 9). 3501 join(o1, u1.orders). 3502 options(joinedload(u1.addresses), 3503 joinedload(o1.items)).filter(u1.id == 7). 3504 filter(Order.id < o1.id). 3505 order_by(User.id, Order.id, u1.id, o1.id).all(), 3506 ) 3507 self.assert_sql_count(testing.db, go, 1) 3508 3509 def test_aliased_entity_one(self): 3510 Item, Order, User, Address = (self.classes.Item, 3511 self.classes.Order, 3512 self.classes.User, 3513 self.classes.Address) 3514 3515 sess = create_session() 3516 3517 oalias = sa.orm.aliased(Order) 3518 3519 # two FROM clauses 3520 def go(): 3521 eq_( 3522 [ 3523 ( 3524 User( 3525 id=9, addresses=[ 3526 Address( 3527 id=5)]), Order( 3528 id=2, items=[ 3529 Item( 3530 id=1), Item( 3531 id=2), Item( 3532 id=3)])), 3533 (User(id=9, addresses=[Address(id=5)]), Order( 3534 id=4, items=[Item(id=1), Item(id=5)])), 3535 ], 3536 sess.query(User, oalias).filter(User.id == oalias.user_id). 3537 options( 3538 joinedload(User.addresses), 3539 joinedload(oalias.items)).filter(User.id == 9). 3540 order_by(User.id, oalias.id).all(), 3541 ) 3542 self.assert_sql_count(testing.db, go, 1) 3543 3544 def test_aliased_entity_two(self): 3545 Item, Order, User, Address = (self.classes.Item, 3546 self.classes.Order, 3547 self.classes.User, 3548 self.classes.Address) 3549 3550 sess = create_session() 3551 3552 oalias = sa.orm.aliased(Order) 3553 3554 # one FROM clause 3555 def go(): 3556 eq_( 3557 [ 3558 ( 3559 User( 3560 id=9, addresses=[ 3561 Address( 3562 id=5)]), Order( 3563 id=2, items=[ 3564 Item( 3565 id=1), Item( 3566 id=2), Item( 3567 id=3)])), 3568 (User(id=9, addresses=[Address(id=5)]), Order( 3569 id=4, items=[Item(id=1), Item(id=5)])), 3570 ], 3571 sess.query(User, oalias).join(oalias, User.orders). 3572 options(joinedload(User.addresses), 3573 joinedload(oalias.items)). 3574 filter(User.id == 9). 3575 order_by(User.id, oalias.id).all(), 3576 ) 3577 self.assert_sql_count(testing.db, go, 1) 3578 3579 def test_aliased_entity_three(self): 3580 Order, User = ( 3581 self.classes.Order, 3582 self.classes.User) 3583 3584 sess = create_session() 3585 3586 oalias = sa.orm.aliased(Order) 3587 3588 # improper setup: oalias in the columns clause but join to usual 3589 # orders alias. this should create two FROM clauses even though the 3590 # query has a from_clause set up via the join 3591 self.assert_compile( 3592 sess.query(User, oalias).join(User.orders). 3593 options(joinedload(oalias.items)).with_labels().statement, 3594 "SELECT users.id AS users_id, users.name AS users_name, " 3595 "orders_1.id AS orders_1_id, " 3596 "orders_1.user_id AS orders_1_user_id, " 3597 "orders_1.address_id AS orders_1_address_id, " 3598 "orders_1.description AS orders_1_description, " 3599 "orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, " 3600 "items_1.description AS items_1_description FROM users " 3601 "JOIN orders ON users.id = orders.user_id, " 3602 "orders AS orders_1 LEFT OUTER JOIN (order_items AS order_items_1 " 3603 "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " 3604 "ON orders_1.id = order_items_1.order_id ORDER BY items_1.id" 3605 ) 3606 3607 3608class SubqueryTest(fixtures.MappedTest): 3609 3610 @classmethod 3611 def define_tables(cls, metadata): 3612 Table('users_table', metadata, 3613 Column( 3614 'id', 3615 Integer, 3616 primary_key=True, 3617 test_needs_autoincrement=True), 3618 Column('name', String(16)) 3619 ) 3620 3621 Table('tags_table', metadata, 3622 Column( 3623 'id', 3624 Integer, 3625 primary_key=True, 3626 test_needs_autoincrement=True), 3627 Column('user_id', Integer, ForeignKey("users_table.id")), 3628 Column('score1', sa.Float), 3629 Column('score2', sa.Float), 3630 ) 3631 3632 def test_label_anonymizing(self): 3633 """Eager loading works with subqueries with labels, 3634 3635 Even if an explicit labelname which conflicts with a label on the 3636 parent. 3637 3638 There's not much reason a column_property() would ever need to have a 3639 label of a specific name (and they don't even need labels these days), 3640 unless you'd like the name to line up with a name that you may be 3641 using for a straight textual statement used for loading instances of 3642 that type. 3643 3644 """ 3645 3646 tags_table, users_table = self.tables.tags_table, \ 3647 self.tables.users_table 3648 3649 class User(fixtures.ComparableEntity): 3650 3651 @property 3652 def prop_score(self): 3653 return sum([tag.prop_score for tag in self.tags]) 3654 3655 class Tag(fixtures.ComparableEntity): 3656 3657 @property 3658 def prop_score(self): 3659 return self.score1 * self.score2 3660 3661 for labeled, labelname in [(True, 'score'), (True, None), 3662 (False, None)]: 3663 sa.orm.clear_mappers() 3664 3665 tag_score = (tags_table.c.score1 * tags_table.c.score2) 3666 user_score = sa.select([sa.func.sum(tags_table.c.score1 * 3667 tags_table.c.score2)], 3668 tags_table.c.user_id == users_table.c.id) 3669 3670 if labeled: 3671 tag_score = tag_score.label(labelname) 3672 user_score = user_score.label(labelname) 3673 else: 3674 user_score = user_score.as_scalar() 3675 3676 mapper(Tag, tags_table, properties={ 3677 'query_score': sa.orm.column_property(tag_score), 3678 }) 3679 3680 mapper(User, users_table, properties={ 3681 'tags': relationship(Tag, backref='user', lazy='joined'), 3682 'query_score': sa.orm.column_property(user_score), 3683 }) 3684 3685 session = create_session() 3686 session.add(User(name='joe', tags=[Tag(score1=5.0, score2=3.0), 3687 Tag(score1=55.0, score2=1.0)])) 3688 session.add(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), 3689 Tag(score1=50.0, score2=1.0), 3690 Tag(score1=15.0, score2=2.0)])) 3691 session.flush() 3692 session.expunge_all() 3693 3694 for user in session.query(User).all(): 3695 eq_(user.query_score, user.prop_score) 3696 3697 def go(): 3698 u = session.query(User).filter_by(name='joe').one() 3699 eq_(u.query_score, u.prop_score) 3700 self.assert_sql_count(testing.db, go, 1) 3701 3702 for t in (tags_table, users_table): 3703 t.delete().execute() 3704 3705 3706class CorrelatedSubqueryTest(fixtures.MappedTest): 3707 3708 """tests for #946, #947, #948. 3709 3710 The "users" table is joined to "stuff", and the relationship 3711 would like to pull only the "stuff" entry with the most recent date. 3712 3713 Exercises a variety of ways to configure this. 3714 3715 """ 3716 3717 # another argument for joinedload learning about inner joins 3718 3719 __requires__ = ('correlated_outer_joins', ) 3720 3721 @classmethod 3722 def define_tables(cls, metadata): 3723 Table( 3724 'users', metadata, 3725 Column( 3726 'id', 3727 Integer, 3728 primary_key=True, 3729 test_needs_autoincrement=True), 3730 Column('name', String(50)) 3731 ) 3732 3733 Table( 3734 'stuff', metadata, 3735 Column( 3736 'id', 3737 Integer, 3738 primary_key=True, 3739 test_needs_autoincrement=True), 3740 Column('date', Date), 3741 Column('user_id', Integer, ForeignKey('users.id'))) 3742 3743 @classmethod 3744 def insert_data(cls): 3745 stuff, users = cls.tables.stuff, cls.tables.users 3746 3747 users.insert().execute( 3748 {'id': 1, 'name': 'user1'}, 3749 {'id': 2, 'name': 'user2'}, 3750 {'id': 3, 'name': 'user3'}, 3751 ) 3752 3753 stuff.insert().execute( 3754 {'id': 1, 'user_id': 1, 'date': datetime.date(2007, 10, 15)}, 3755 {'id': 2, 'user_id': 1, 'date': datetime.date(2007, 12, 15)}, 3756 {'id': 3, 'user_id': 1, 'date': datetime.date(2007, 11, 15)}, 3757 {'id': 4, 'user_id': 2, 'date': datetime.date(2008, 1, 15)}, 3758 {'id': 5, 'user_id': 3, 'date': datetime.date(2007, 6, 15)}, 3759 {'id': 6, 'user_id': 3, 'date': datetime.date(2007, 3, 15)}, 3760 ) 3761 3762 def test_labeled_on_date_noalias(self): 3763 self._do_test('label', True, False) 3764 3765 def test_scalar_on_date_noalias(self): 3766 self._do_test('scalar', True, False) 3767 3768 def test_plain_on_date_noalias(self): 3769 self._do_test('none', True, False) 3770 3771 def test_labeled_on_limitid_noalias(self): 3772 self._do_test('label', False, False) 3773 3774 def test_scalar_on_limitid_noalias(self): 3775 self._do_test('scalar', False, False) 3776 3777 def test_plain_on_limitid_noalias(self): 3778 self._do_test('none', False, False) 3779 3780 def test_labeled_on_date_alias(self): 3781 self._do_test('label', True, True) 3782 3783 def test_scalar_on_date_alias(self): 3784 self._do_test('scalar', True, True) 3785 3786 def test_plain_on_date_alias(self): 3787 self._do_test('none', True, True) 3788 3789 def test_labeled_on_limitid_alias(self): 3790 self._do_test('label', False, True) 3791 3792 def test_scalar_on_limitid_alias(self): 3793 self._do_test('scalar', False, True) 3794 3795 def test_plain_on_limitid_alias(self): 3796 self._do_test('none', False, True) 3797 3798 def _do_test(self, labeled, ondate, aliasstuff): 3799 stuff, users = self.tables.stuff, self.tables.users 3800 3801 class User(fixtures.ComparableEntity): 3802 pass 3803 3804 class Stuff(fixtures.ComparableEntity): 3805 pass 3806 3807 mapper(Stuff, stuff) 3808 3809 if aliasstuff: 3810 salias = stuff.alias() 3811 else: 3812 # if we don't alias the 'stuff' table within the correlated 3813 # subquery, 3814 # it gets aliased in the eager load along with the "stuff" table 3815 # to "stuff_1". 3816 # but it's a scalar subquery, and this doesn't actually matter 3817 salias = stuff 3818 3819 if ondate: 3820 # the more 'relational' way to do this, join on the max date 3821 stuff_view = select([func.max(salias.c.date).label('max_date')]).\ 3822 where(salias.c.user_id == users.c.id).correlate(users) 3823 else: 3824 # a common method with the MySQL crowd, which actually might 3825 # perform better in some 3826 # cases - subquery does a limit with order by DESC, join on the id 3827 stuff_view = select([salias.c.id]).\ 3828 where(salias.c.user_id == users.c.id).\ 3829 correlate(users).order_by(salias.c.date.desc()).limit(1) 3830 3831 # can't win on this one 3832 if testing.against("mssql"): 3833 operator = operators.in_op 3834 else: 3835 operator = operators.eq 3836 3837 if labeled == 'label': 3838 stuff_view = stuff_view.label('foo') 3839 operator = operators.eq 3840 elif labeled == 'scalar': 3841 stuff_view = stuff_view.as_scalar() 3842 3843 if ondate: 3844 mapper(User, users, properties={ 3845 'stuff': relationship( 3846 Stuff, 3847 primaryjoin=and_(users.c.id == stuff.c.user_id, 3848 operator(stuff.c.date, stuff_view))) 3849 }) 3850 else: 3851 mapper(User, users, properties={ 3852 'stuff': relationship( 3853 Stuff, 3854 primaryjoin=and_(users.c.id == stuff.c.user_id, 3855 operator(stuff.c.id, stuff_view))) 3856 }) 3857 3858 sess = create_session() 3859 3860 def go(): 3861 eq_( 3862 sess.query(User).order_by(User.name).options( 3863 joinedload('stuff')).all(), 3864 [ 3865 User(name='user1', stuff=[Stuff(id=2)]), 3866 User(name='user2', stuff=[Stuff(id=4)]), 3867 User(name='user3', stuff=[Stuff(id=5)]) 3868 ] 3869 ) 3870 self.assert_sql_count(testing.db, go, 1) 3871 3872 sess = create_session() 3873 3874 def go(): 3875 eq_( 3876 sess.query(User).order_by(User.name).first(), 3877 User(name='user1', stuff=[Stuff(id=2)]) 3878 ) 3879 self.assert_sql_count(testing.db, go, 2) 3880 3881 sess = create_session() 3882 3883 def go(): 3884 eq_( 3885 sess.query(User).order_by(User.name).options( 3886 joinedload('stuff')).first(), 3887 User(name='user1', stuff=[Stuff(id=2)]) 3888 ) 3889 self.assert_sql_count(testing.db, go, 1) 3890 3891 sess = create_session() 3892 3893 def go(): 3894 eq_( 3895 sess.query(User).filter(User.id == 2).options( 3896 joinedload('stuff')).one(), 3897 User(name='user2', stuff=[Stuff(id=4)]) 3898 ) 3899 self.assert_sql_count(testing.db, go, 1) 3900 3901 3902class CyclicalInheritingEagerTestOne(fixtures.MappedTest): 3903 3904 @classmethod 3905 def define_tables(cls, metadata): 3906 Table( 3907 't1', metadata, 3908 Column( 3909 'c1', Integer, primary_key=True, 3910 test_needs_autoincrement=True), 3911 Column('c2', String(30)), 3912 Column('type', String(30)) 3913 ) 3914 3915 Table('t2', metadata, 3916 Column('c1', Integer, primary_key=True, 3917 test_needs_autoincrement=True), 3918 Column('c2', String(30)), 3919 Column('type', String(30)), 3920 Column('t1.id', Integer, ForeignKey('t1.c1'))) 3921 3922 def test_basic(self): 3923 t2, t1 = self.tables.t2, self.tables.t1 3924 3925 class T(object): 3926 pass 3927 3928 class SubT(T): 3929 pass 3930 3931 class T2(object): 3932 pass 3933 3934 class SubT2(T2): 3935 pass 3936 3937 mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1') 3938 mapper( 3939 SubT, None, inherits=T, polymorphic_identity='subt1', 3940 properties={ 3941 't2s': relationship( 3942 SubT2, lazy='joined', 3943 backref=sa.orm.backref('subt', lazy='joined')) 3944 }) 3945 mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2') 3946 mapper(SubT2, None, inherits=T2, polymorphic_identity='subt2') 3947 3948 # testing a particular endless loop condition in eager load setup 3949 create_session().query(SubT).all() 3950 3951 3952class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest, 3953 testing.AssertsCompiledSQL): 3954 __dialect__ = 'default' 3955 3956 @classmethod 3957 def setup_classes(cls): 3958 Base = cls.DeclarativeBasic 3959 3960 class PersistentObject(Base): 3961 __tablename__ = 'persistent' 3962 id = Column(Integer, primary_key=True, 3963 test_needs_autoincrement=True) 3964 3965 class Movie(PersistentObject): 3966 __tablename__ = 'movie' 3967 id = Column(Integer, ForeignKey('persistent.id'), primary_key=True) 3968 director_id = Column(Integer, ForeignKey('director.id')) 3969 title = Column(String(50)) 3970 3971 class Director(PersistentObject): 3972 __tablename__ = 'director' 3973 id = Column(Integer, ForeignKey('persistent.id'), primary_key=True) 3974 movies = relationship("Movie", foreign_keys=Movie.director_id) 3975 name = Column(String(50)) 3976 3977 def test_from_subclass(self): 3978 Director = self.classes.Director 3979 s = create_session() 3980 3981 self.assert_compile( 3982 s.query(Director).options(joinedload('*')), 3983 "SELECT director.id AS director_id, " 3984 "persistent.id AS persistent_id, " 3985 "director.name AS director_name, movie_1.id AS movie_1_id, " 3986 "persistent_1.id AS persistent_1_id, " 3987 "movie_1.director_id AS movie_1_director_id, " 3988 "movie_1.title AS movie_1_title " 3989 "FROM persistent JOIN director ON persistent.id = director.id " 3990 "LEFT OUTER JOIN " 3991 "(persistent AS persistent_1 JOIN movie AS movie_1 " 3992 "ON persistent_1.id = movie_1.id) " 3993 "ON director.id = movie_1.director_id" 3994 ) 3995 3996 def test_integrate(self): 3997 Director = self.classes.Director 3998 Movie = self.classes.Movie 3999 4000 session = Session(testing.db) 4001 rscott = Director(name="Ridley Scott") 4002 alien = Movie(title="Alien") 4003 brunner = Movie(title="Blade Runner") 4004 rscott.movies.append(brunner) 4005 rscott.movies.append(alien) 4006 session.add_all([rscott, alien, brunner]) 4007 session.commit() 4008 4009 session.close_all() 4010 self.d = session.query(Director).options(joinedload('*')).first() 4011 assert len(list(session)) == 3 4012 4013 4014class CyclicalInheritingEagerTestThree(fixtures.DeclarativeMappedTest, 4015 testing.AssertsCompiledSQL): 4016 __dialect__ = 'default' 4017 run_create_tables = None 4018 4019 @classmethod 4020 def setup_classes(cls): 4021 Base = cls.DeclarativeBasic 4022 4023 class PersistentObject(Base): 4024 __tablename__ = 'persistent' 4025 id = Column(Integer, primary_key=True, 4026 test_needs_autoincrement=True) 4027 4028 __mapper_args__ = {'with_polymorphic': "*"} 4029 4030 class Director(PersistentObject): 4031 __tablename__ = 'director' 4032 id = Column(Integer, ForeignKey('persistent.id'), primary_key=True) 4033 other_id = Column(Integer, ForeignKey('persistent.id')) 4034 name = Column(String(50)) 4035 other = relationship(PersistentObject, 4036 primaryjoin=other_id == PersistentObject.id, 4037 lazy=False) 4038 __mapper_args__ = {"inherit_condition": id == PersistentObject.id} 4039 4040 def test_gen_query_nodepth(self): 4041 PersistentObject = self.classes.PersistentObject 4042 sess = create_session() 4043 self.assert_compile( 4044 sess.query(PersistentObject), 4045 "SELECT persistent.id AS persistent_id, " 4046 "director.id AS director_id," 4047 " director.other_id AS director_other_id, " 4048 "director.name AS director_name FROM persistent " 4049 "LEFT OUTER JOIN director ON director.id = persistent.id" 4050 ) 4051 4052 def test_gen_query_depth(self): 4053 PersistentObject = self.classes.PersistentObject 4054 Director = self.classes.Director 4055 sess = create_session() 4056 self.assert_compile( 4057 sess.query(PersistentObject).options(joinedload(Director.other)), 4058 "SELECT persistent.id AS persistent_id, " 4059 "director.id AS director_id, " 4060 "director.other_id AS director_other_id, " 4061 "director.name AS director_name, persistent_1.id AS " 4062 "persistent_1_id, director_1.id AS director_1_id, " 4063 "director_1.other_id AS director_1_other_id, " 4064 "director_1.name AS director_1_name " 4065 "FROM persistent LEFT OUTER JOIN director " 4066 "ON director.id = persistent.id " 4067 "LEFT OUTER JOIN (persistent AS persistent_1 " 4068 "LEFT OUTER JOIN director AS director_1 ON " 4069 "director_1.id = persistent_1.id) " 4070 "ON director.other_id = persistent_1.id" 4071 ) 4072 4073 4074class EnsureColumnsAddedTest( 4075 fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL): 4076 __dialect__ = 'default' 4077 run_create_tables = None 4078 4079 @classmethod 4080 def setup_classes(cls): 4081 Base = cls.DeclarativeBasic 4082 4083 class Parent(Base): 4084 __tablename__ = 'parent' 4085 id = Column(Integer, primary_key=True, 4086 test_needs_autoincrement=True) 4087 arb = Column(Integer, unique=True) 4088 data = Column(Integer) 4089 o2mchild = relationship("O2MChild") 4090 m2mchild = relationship("M2MChild", secondary=Table( 4091 'parent_to_m2m', Base.metadata, 4092 Column('parent_id', ForeignKey('parent.arb')), 4093 Column('child_id', ForeignKey('m2mchild.id')) 4094 )) 4095 4096 class O2MChild(Base): 4097 __tablename__ = 'o2mchild' 4098 id = Column(Integer, primary_key=True, 4099 test_needs_autoincrement=True) 4100 parent_id = Column(ForeignKey('parent.arb')) 4101 4102 class M2MChild(Base): 4103 __tablename__ = 'm2mchild' 4104 id = Column(Integer, primary_key=True, 4105 test_needs_autoincrement=True) 4106 4107 def test_joinedload_defered_pk_limit_o2m(self): 4108 Parent = self.classes.Parent 4109 4110 s = Session() 4111 4112 self.assert_compile( 4113 s.query(Parent).options( 4114 load_only('data'), 4115 joinedload(Parent.o2mchild)).limit(10), 4116 "SELECT anon_1.parent_id AS anon_1_parent_id, " 4117 "anon_1.parent_data AS anon_1_parent_data, " 4118 "anon_1.parent_arb AS anon_1_parent_arb, " 4119 "o2mchild_1.id AS o2mchild_1_id, " 4120 "o2mchild_1.parent_id AS o2mchild_1_parent_id " 4121 "FROM (SELECT parent.id AS parent_id, parent.data AS parent_data, " 4122 "parent.arb AS parent_arb FROM parent LIMIT :param_1) AS anon_1 " 4123 "LEFT OUTER JOIN o2mchild AS o2mchild_1 " 4124 "ON anon_1.parent_arb = o2mchild_1.parent_id" 4125 ) 4126 4127 def test_joinedload_defered_pk_limit_m2m(self): 4128 Parent = self.classes.Parent 4129 4130 s = Session() 4131 4132 self.assert_compile( 4133 s.query(Parent).options( 4134 load_only('data'), 4135 joinedload(Parent.m2mchild)).limit(10), 4136 "SELECT anon_1.parent_id AS anon_1_parent_id, " 4137 "anon_1.parent_data AS anon_1_parent_data, " 4138 "anon_1.parent_arb AS anon_1_parent_arb, " 4139 "m2mchild_1.id AS m2mchild_1_id " 4140 "FROM (SELECT parent.id AS parent_id, " 4141 "parent.data AS parent_data, parent.arb AS parent_arb " 4142 "FROM parent LIMIT :param_1) AS anon_1 " 4143 "LEFT OUTER JOIN (parent_to_m2m AS parent_to_m2m_1 " 4144 "JOIN m2mchild AS m2mchild_1 " 4145 "ON m2mchild_1.id = parent_to_m2m_1.child_id) " 4146 "ON anon_1.parent_arb = parent_to_m2m_1.parent_id" 4147 ) 4148 4149 def test_joinedload_defered_pk_o2m(self): 4150 Parent = self.classes.Parent 4151 4152 s = Session() 4153 4154 self.assert_compile( 4155 s.query(Parent).options( 4156 load_only('data'), 4157 joinedload(Parent.o2mchild)), 4158 "SELECT parent.id AS parent_id, parent.data AS parent_data, " 4159 "parent.arb AS parent_arb, o2mchild_1.id AS o2mchild_1_id, " 4160 "o2mchild_1.parent_id AS o2mchild_1_parent_id " 4161 "FROM parent LEFT OUTER JOIN o2mchild AS o2mchild_1 " 4162 "ON parent.arb = o2mchild_1.parent_id" 4163 ) 4164 4165 def test_joinedload_defered_pk_m2m(self): 4166 Parent = self.classes.Parent 4167 4168 s = Session() 4169 4170 self.assert_compile( 4171 s.query(Parent).options( 4172 load_only('data'), 4173 joinedload(Parent.m2mchild)), 4174 "SELECT parent.id AS parent_id, parent.data AS parent_data, " 4175 "parent.arb AS parent_arb, m2mchild_1.id AS m2mchild_1_id " 4176 "FROM parent LEFT OUTER JOIN (parent_to_m2m AS parent_to_m2m_1 " 4177 "JOIN m2mchild AS m2mchild_1 " 4178 "ON m2mchild_1.id = parent_to_m2m_1.child_id) " 4179 "ON parent.arb = parent_to_m2m_1.parent_id" 4180 ) 4181