1import sqlalchemy as sa 2from sqlalchemy import testing, util 3from sqlalchemy.orm import mapper, deferred, defer, undefer, Load, \ 4 load_only, undefer_group, create_session, synonym, relationship, Session,\ 5 joinedload, defaultload, aliased, contains_eager, with_polymorphic 6from sqlalchemy.testing import eq_, AssertsCompiledSQL, assert_raises_message 7from test.orm import _fixtures 8 9 10from .inheritance._poly_fixtures import Company, Person, Engineer, Manager, \ 11 Boss, Machine, Paperwork, _Polymorphic 12 13 14class DeferredTest(AssertsCompiledSQL, _fixtures.FixtureTest): 15 16 def test_basic(self): 17 """A basic deferred load.""" 18 19 Order, orders = self.classes.Order, self.tables.orders 20 21 22 mapper(Order, orders, order_by=orders.c.id, properties={ 23 'description': deferred(orders.c.description)}) 24 25 o = Order() 26 self.assert_(o.description is None) 27 28 q = create_session().query(Order) 29 def go(): 30 l = q.all() 31 o2 = l[2] 32 x = o2.description 33 34 self.sql_eq_(go, [ 35 ("SELECT orders.id AS orders_id, " 36 "orders.user_id AS orders_user_id, " 37 "orders.address_id AS orders_address_id, " 38 "orders.isopen AS orders_isopen " 39 "FROM orders ORDER BY orders.id", {}), 40 ("SELECT orders.description AS orders_description " 41 "FROM orders WHERE orders.id = :param_1", 42 {'param_1':3})]) 43 44 def test_defer_primary_key(self): 45 """what happens when we try to defer the primary key?""" 46 47 Order, orders = self.classes.Order, self.tables.orders 48 49 50 mapper(Order, orders, order_by=orders.c.id, properties={ 51 'id': deferred(orders.c.id)}) 52 53 # right now, it's not that graceful :) 54 q = create_session().query(Order) 55 assert_raises_message( 56 sa.exc.NoSuchColumnError, 57 "Could not locate", 58 q.first 59 ) 60 61 62 def test_unsaved(self): 63 """Deferred loading does not kick in when just PK cols are set.""" 64 65 Order, orders = self.classes.Order, self.tables.orders 66 67 68 mapper(Order, orders, properties={ 69 'description': deferred(orders.c.description)}) 70 71 sess = create_session() 72 o = Order() 73 sess.add(o) 74 o.id = 7 75 def go(): 76 o.description = "some description" 77 self.sql_count_(0, go) 78 79 def test_synonym_group_bug(self): 80 orders, Order = self.tables.orders, self.classes.Order 81 82 mapper(Order, orders, properties={ 83 'isopen':synonym('_isopen', map_column=True), 84 'description':deferred(orders.c.description, group='foo') 85 }) 86 87 sess = create_session() 88 o1 = sess.query(Order).get(1) 89 eq_(o1.description, "order 1") 90 91 def test_unsaved_2(self): 92 Order, orders = self.classes.Order, self.tables.orders 93 94 mapper(Order, orders, properties={ 95 'description': deferred(orders.c.description)}) 96 97 sess = create_session() 98 o = Order() 99 sess.add(o) 100 def go(): 101 o.description = "some description" 102 self.sql_count_(0, go) 103 104 def test_unsaved_group(self): 105 """Deferred loading doesn't kick in when just PK cols are set""" 106 107 orders, Order = self.tables.orders, self.classes.Order 108 109 110 mapper(Order, orders, order_by=orders.c.id, properties=dict( 111 description=deferred(orders.c.description, group='primary'), 112 opened=deferred(orders.c.isopen, group='primary'))) 113 114 sess = create_session() 115 o = Order() 116 sess.add(o) 117 o.id = 7 118 def go(): 119 o.description = "some description" 120 self.sql_count_(0, go) 121 122 def test_unsaved_group_2(self): 123 orders, Order = self.tables.orders, self.classes.Order 124 125 mapper(Order, orders, order_by=orders.c.id, properties=dict( 126 description=deferred(orders.c.description, group='primary'), 127 opened=deferred(orders.c.isopen, group='primary'))) 128 129 sess = create_session() 130 o = Order() 131 sess.add(o) 132 def go(): 133 o.description = "some description" 134 self.sql_count_(0, go) 135 136 def test_save(self): 137 Order, orders = self.classes.Order, self.tables.orders 138 139 m = mapper(Order, orders, properties={ 140 'description': deferred(orders.c.description)}) 141 142 sess = create_session() 143 o2 = sess.query(Order).get(2) 144 o2.isopen = 1 145 sess.flush() 146 147 def test_group(self): 148 """Deferred load with a group""" 149 150 orders, Order = self.tables.orders, self.classes.Order 151 152 mapper(Order, orders, properties=util.OrderedDict([ 153 ('userident', deferred(orders.c.user_id, group='primary')), 154 ('addrident', deferred(orders.c.address_id, group='primary')), 155 ('description', deferred(orders.c.description, group='primary')), 156 ('opened', deferred(orders.c.isopen, group='primary')) 157 ])) 158 159 sess = create_session() 160 q = sess.query(Order).order_by(Order.id) 161 def go(): 162 l = q.all() 163 o2 = l[2] 164 eq_(o2.opened, 1) 165 eq_(o2.userident, 7) 166 eq_(o2.description, 'order 3') 167 168 self.sql_eq_(go, [ 169 ("SELECT orders.id AS orders_id " 170 "FROM orders ORDER BY orders.id", {}), 171 ("SELECT orders.user_id AS orders_user_id, " 172 "orders.address_id AS orders_address_id, " 173 "orders.description AS orders_description, " 174 "orders.isopen AS orders_isopen " 175 "FROM orders WHERE orders.id = :param_1", 176 {'param_1':3})]) 177 178 o2 = q.all()[2] 179 eq_(o2.description, 'order 3') 180 assert o2 not in sess.dirty 181 o2.description = 'order 3' 182 def go(): 183 sess.flush() 184 self.sql_count_(0, go) 185 186 def test_preserve_changes(self): 187 """A deferred load operation doesn't revert modifications on attributes""" 188 189 orders, Order = self.tables.orders, self.classes.Order 190 191 mapper(Order, orders, properties = { 192 'userident': deferred(orders.c.user_id, group='primary'), 193 'description': deferred(orders.c.description, group='primary'), 194 'opened': deferred(orders.c.isopen, group='primary') 195 }) 196 sess = create_session() 197 o = sess.query(Order).get(3) 198 assert 'userident' not in o.__dict__ 199 o.description = 'somenewdescription' 200 eq_(o.description, 'somenewdescription') 201 def go(): 202 eq_(o.opened, 1) 203 self.assert_sql_count(testing.db, go, 1) 204 eq_(o.description, 'somenewdescription') 205 assert o in sess.dirty 206 207 def test_commits_state(self): 208 """ 209 When deferred elements are loaded via a group, they get the proper 210 CommittedState and don't result in changes being committed 211 212 """ 213 214 orders, Order = self.tables.orders, self.classes.Order 215 216 mapper(Order, orders, properties = { 217 'userident': deferred(orders.c.user_id, group='primary'), 218 'description': deferred(orders.c.description, group='primary'), 219 'opened': deferred(orders.c.isopen, group='primary')}) 220 221 sess = create_session() 222 o2 = sess.query(Order).get(3) 223 224 # this will load the group of attributes 225 eq_(o2.description, 'order 3') 226 assert o2 not in sess.dirty 227 # this will mark it as 'dirty', but nothing actually changed 228 o2.description = 'order 3' 229 # therefore the flush() shouldn't actually issue any SQL 230 self.assert_sql_count(testing.db, sess.flush, 0) 231 232 def test_map_selectable_wo_deferred(self): 233 """test mapping to a selectable with deferred cols, 234 the selectable doesn't include the deferred col. 235 236 """ 237 238 Order, orders = self.classes.Order, self.tables.orders 239 240 241 order_select = sa.select([ 242 orders.c.id, 243 orders.c.user_id, 244 orders.c.address_id, 245 orders.c.description, 246 orders.c.isopen]).alias() 247 mapper(Order, order_select, properties={ 248 'description':deferred(order_select.c.description) 249 }) 250 251 sess = Session() 252 o1 = sess.query(Order).order_by(Order.id).first() 253 assert 'description' not in o1.__dict__ 254 eq_(o1.description, 'order 1') 255 256 257class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest): 258 __dialect__ = 'default' 259 260 def test_options(self): 261 """Options on a mapper to create deferred and undeferred columns""" 262 263 orders, Order = self.tables.orders, self.classes.Order 264 265 266 mapper(Order, orders) 267 268 sess = create_session() 269 q = sess.query(Order).order_by(Order.id).options(defer('user_id')) 270 271 def go(): 272 q.all()[0].user_id 273 274 self.sql_eq_(go, [ 275 ("SELECT orders.id AS orders_id, " 276 "orders.address_id AS orders_address_id, " 277 "orders.description AS orders_description, " 278 "orders.isopen AS orders_isopen " 279 "FROM orders ORDER BY orders.id", {}), 280 ("SELECT orders.user_id AS orders_user_id " 281 "FROM orders WHERE orders.id = :param_1", 282 {'param_1':1})]) 283 sess.expunge_all() 284 285 q2 = q.options(undefer('user_id')) 286 self.sql_eq_(q2.all, [ 287 ("SELECT orders.id AS orders_id, " 288 "orders.user_id AS orders_user_id, " 289 "orders.address_id AS orders_address_id, " 290 "orders.description AS orders_description, " 291 "orders.isopen AS orders_isopen " 292 "FROM orders ORDER BY orders.id", 293 {})]) 294 295 def test_undefer_group(self): 296 orders, Order = self.tables.orders, self.classes.Order 297 298 mapper(Order, orders, properties=util.OrderedDict([ 299 ('userident', deferred(orders.c.user_id, group='primary')), 300 ('description', deferred(orders.c.description, group='primary')), 301 ('opened', deferred(orders.c.isopen, group='primary')) 302 ] 303 )) 304 305 sess = create_session() 306 q = sess.query(Order).order_by(Order.id) 307 def go(): 308 l = q.options(undefer_group('primary')).all() 309 o2 = l[2] 310 eq_(o2.opened, 1) 311 eq_(o2.userident, 7) 312 eq_(o2.description, 'order 3') 313 314 self.sql_eq_(go, [ 315 ("SELECT orders.user_id AS orders_user_id, " 316 "orders.description AS orders_description, " 317 "orders.isopen AS orders_isopen, " 318 "orders.id AS orders_id, " 319 "orders.address_id AS orders_address_id " 320 "FROM orders ORDER BY orders.id", 321 {})]) 322 323 def test_undefer_group_multi(self): 324 orders, Order = self.tables.orders, self.classes.Order 325 326 mapper(Order, orders, properties=util.OrderedDict([ 327 ('userident', deferred(orders.c.user_id, group='primary')), 328 ('description', deferred(orders.c.description, group='primary')), 329 ('opened', deferred(orders.c.isopen, group='secondary')) 330 ] 331 )) 332 333 sess = create_session() 334 q = sess.query(Order).order_by(Order.id) 335 def go(): 336 l = q.options( 337 undefer_group('primary'), undefer_group('secondary')).all() 338 o2 = l[2] 339 eq_(o2.opened, 1) 340 eq_(o2.userident, 7) 341 eq_(o2.description, 'order 3') 342 343 self.sql_eq_(go, [ 344 ("SELECT orders.user_id AS orders_user_id, " 345 "orders.description AS orders_description, " 346 "orders.isopen AS orders_isopen, " 347 "orders.id AS orders_id, " 348 "orders.address_id AS orders_address_id " 349 "FROM orders ORDER BY orders.id", 350 {})]) 351 352 def test_undefer_group_multi_pathed(self): 353 orders, Order = self.tables.orders, self.classes.Order 354 355 mapper(Order, orders, properties=util.OrderedDict([ 356 ('userident', deferred(orders.c.user_id, group='primary')), 357 ('description', deferred(orders.c.description, group='primary')), 358 ('opened', deferred(orders.c.isopen, group='secondary')) 359 ] 360 )) 361 362 sess = create_session() 363 q = sess.query(Order).order_by(Order.id) 364 def go(): 365 l = q.options( 366 Load(Order).undefer_group('primary').undefer_group('secondary')).all() 367 o2 = l[2] 368 eq_(o2.opened, 1) 369 eq_(o2.userident, 7) 370 eq_(o2.description, 'order 3') 371 372 self.sql_eq_(go, [ 373 ("SELECT orders.user_id AS orders_user_id, " 374 "orders.description AS orders_description, " 375 "orders.isopen AS orders_isopen, " 376 "orders.id AS orders_id, " 377 "orders.address_id AS orders_address_id " 378 "FROM orders ORDER BY orders.id", 379 {})]) 380 381 def test_undefer_star(self): 382 orders, Order = self.tables.orders, self.classes.Order 383 384 mapper(Order, orders, properties=util.OrderedDict([ 385 ('userident', deferred(orders.c.user_id)), 386 ('description', deferred(orders.c.description)), 387 ('opened', deferred(orders.c.isopen)) 388 ] 389 )) 390 391 sess = create_session() 392 q = sess.query(Order).options(Load(Order).undefer('*')) 393 self.assert_compile(q, 394 "SELECT orders.user_id AS orders_user_id, " 395 "orders.description AS orders_description, " 396 "orders.isopen AS orders_isopen, " 397 "orders.id AS orders_id, " 398 "orders.address_id AS orders_address_id FROM orders" 399 ) 400 401 def test_locates_col(self): 402 """changed in 1.0 - we don't search for deferred cols in the result 403 now. """ 404 405 orders, Order = self.tables.orders, self.classes.Order 406 407 408 mapper(Order, orders, properties={ 409 'description': deferred(orders.c.description)}) 410 411 sess = create_session() 412 o1 = (sess.query(Order). 413 order_by(Order.id). 414 add_column(orders.c.description).first())[0] 415 def go(): 416 eq_(o1.description, 'order 1') 417 # prior to 1.0 we'd search in the result for this column 418 # self.sql_count_(0, go) 419 self.sql_count_(1, go) 420 421 def test_locates_col_rowproc_only(self): 422 """changed in 1.0 - we don't search for deferred cols in the result 423 now. 424 425 Because the loading for ORM Query and Query from a core select 426 is now split off, we test loading from a plain select() 427 separately. 428 429 """ 430 431 orders, Order = self.tables.orders, self.classes.Order 432 433 434 mapper(Order, orders, properties={ 435 'description': deferred(orders.c.description)}) 436 437 sess = create_session() 438 stmt = sa.select([Order]).order_by(Order.id) 439 o1 = (sess.query(Order). 440 from_statement(stmt).all())[0] 441 def go(): 442 eq_(o1.description, 'order 1') 443 # prior to 1.0 we'd search in the result for this column 444 # self.sql_count_(0, go) 445 self.sql_count_(1, go) 446 447 def test_deep_options(self): 448 users, items, order_items, Order, Item, User, orders = (self.tables.users, 449 self.tables.items, 450 self.tables.order_items, 451 self.classes.Order, 452 self.classes.Item, 453 self.classes.User, 454 self.tables.orders) 455 456 mapper(Item, items, properties=dict( 457 description=deferred(items.c.description))) 458 mapper(Order, orders, properties=dict( 459 items=relationship(Item, secondary=order_items))) 460 mapper(User, users, properties=dict( 461 orders=relationship(Order, order_by=orders.c.id))) 462 463 sess = create_session() 464 q = sess.query(User).order_by(User.id) 465 l = q.all() 466 item = l[0].orders[1].items[1] 467 def go(): 468 eq_(item.description, 'item 4') 469 self.sql_count_(1, go) 470 eq_(item.description, 'item 4') 471 472 sess.expunge_all() 473 l = q.options(undefer('orders.items.description')).all() 474 item = l[0].orders[1].items[1] 475 def go(): 476 eq_(item.description, 'item 4') 477 self.sql_count_(0, go) 478 eq_(item.description, 'item 4') 479 480 def test_path_entity(self): 481 """test the legacy *addl_attrs argument.""" 482 483 User = self.classes.User 484 Order = self.classes.Order 485 Item = self.classes.Item 486 487 users = self.tables.users 488 orders = self.tables.orders 489 items = self.tables.items 490 order_items = self.tables.order_items 491 492 mapper(User, users, properties={ 493 "orders": relationship(Order, lazy="joined") 494 }) 495 mapper(Order, orders, properties={ 496 "items": relationship(Item, secondary=order_items, lazy="joined") 497 }) 498 mapper(Item, items) 499 500 sess = create_session() 501 502 exp = ("SELECT users.id AS users_id, users.name AS users_name, " 503 "items_1.id AS items_1_id, orders_1.id AS orders_1_id, " 504 "orders_1.user_id AS orders_1_user_id, orders_1.address_id " 505 "AS orders_1_address_id, orders_1.description AS " 506 "orders_1_description, orders_1.isopen AS orders_1_isopen " 507 "FROM users LEFT OUTER JOIN orders AS orders_1 " 508 "ON users.id = orders_1.user_id LEFT OUTER JOIN " 509 "(order_items AS order_items_1 JOIN items AS items_1 " 510 "ON items_1.id = order_items_1.item_id) " 511 "ON orders_1.id = order_items_1.order_id") 512 513 q = sess.query(User).options(defer(User.orders, Order.items, Item.description)) 514 self.assert_compile(q, exp) 515 516 517 def test_chained_multi_col_options(self): 518 users, User = self.tables.users, self.classes.User 519 orders, Order = self.tables.orders, self.classes.Order 520 521 mapper(User, users, properties={ 522 "orders": relationship(Order) 523 }) 524 mapper(Order, orders) 525 526 sess = create_session() 527 q = sess.query(User).options( 528 joinedload(User.orders).defer("description").defer("isopen") 529 ) 530 self.assert_compile(q, 531 "SELECT users.id AS users_id, users.name AS users_name, " 532 "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, " 533 "orders_1.address_id AS orders_1_address_id FROM users " 534 "LEFT OUTER JOIN orders AS orders_1 ON users.id = orders_1.user_id" 535 ) 536 537 def test_load_only_no_pk(self): 538 orders, Order = self.tables.orders, self.classes.Order 539 540 mapper(Order, orders) 541 542 sess = create_session() 543 q = sess.query(Order).options(load_only("isopen", "description")) 544 self.assert_compile(q, 545 "SELECT orders.id AS orders_id, " 546 "orders.description AS orders_description, " 547 "orders.isopen AS orders_isopen FROM orders") 548 549 def test_load_only_no_pk_rt(self): 550 orders, Order = self.tables.orders, self.classes.Order 551 552 mapper(Order, orders) 553 554 sess = create_session() 555 q = sess.query(Order).order_by(Order.id).\ 556 options(load_only("isopen", "description")) 557 eq_(q.first(), Order(id=1)) 558 559 def test_load_only_w_deferred(self): 560 orders, Order = self.tables.orders, self.classes.Order 561 562 mapper(Order, orders, properties={ 563 "description": deferred(orders.c.description) 564 }) 565 566 sess = create_session() 567 q = sess.query(Order).options( 568 load_only("isopen", "description"), 569 undefer("user_id") 570 ) 571 self.assert_compile(q, 572 "SELECT orders.description AS orders_description, " 573 "orders.id AS orders_id, " 574 "orders.user_id AS orders_user_id, " 575 "orders.isopen AS orders_isopen FROM orders") 576 577 def test_load_only_propagate_unbound(self): 578 self._test_load_only_propagate(False) 579 580 def test_load_only_propagate_bound(self): 581 self._test_load_only_propagate(True) 582 583 def _test_load_only_propagate(self, use_load): 584 User = self.classes.User 585 Address = self.classes.Address 586 587 users = self.tables.users 588 addresses = self.tables.addresses 589 590 mapper(User, users, properties={ 591 "addresses": relationship(Address) 592 }) 593 mapper(Address, addresses) 594 595 sess = create_session() 596 expected = [ 597 ("SELECT users.id AS users_id, users.name AS users_name " 598 "FROM users WHERE users.id IN (:id_1, :id_2)", {'id_2': 8, 'id_1': 7}), 599 ("SELECT addresses.id AS addresses_id, " 600 "addresses.email_address AS addresses_email_address " 601 "FROM addresses WHERE :param_1 = addresses.user_id", {'param_1': 7}), 602 ("SELECT addresses.id AS addresses_id, " 603 "addresses.email_address AS addresses_email_address " 604 "FROM addresses WHERE :param_1 = addresses.user_id", {'param_1': 8}), 605 ] 606 607 if use_load: 608 opt = Load(User).defaultload(User.addresses).load_only("id", "email_address") 609 else: 610 opt = defaultload(User.addresses).load_only("id", "email_address") 611 q = sess.query(User).options(opt).filter(User.id.in_([7, 8])) 612 def go(): 613 for user in q: 614 user.addresses 615 616 self.sql_eq_(go, expected) 617 618 619 def test_load_only_parent_specific(self): 620 User = self.classes.User 621 Address = self.classes.Address 622 Order = self.classes.Order 623 624 users = self.tables.users 625 addresses = self.tables.addresses 626 orders = self.tables.orders 627 628 mapper(User, users) 629 mapper(Address, addresses) 630 mapper(Order, orders) 631 632 sess = create_session() 633 q = sess.query(User, Order, Address).options( 634 Load(User).load_only("name"), 635 Load(Order).load_only("id"), 636 Load(Address).load_only("id", "email_address") 637 ) 638 639 self.assert_compile(q, 640 "SELECT users.id AS users_id, users.name AS users_name, " 641 "orders.id AS orders_id, " 642 "addresses.id AS addresses_id, addresses.email_address " 643 "AS addresses_email_address FROM users, orders, addresses" 644 ) 645 646 def test_load_only_path_specific(self): 647 User = self.classes.User 648 Address = self.classes.Address 649 Order = self.classes.Order 650 651 users = self.tables.users 652 addresses = self.tables.addresses 653 orders = self.tables.orders 654 655 mapper(User, users, properties=util.OrderedDict([ 656 ("addresses", relationship(Address, lazy="joined")), 657 ("orders", relationship(Order, lazy="joined")) 658 ])) 659 660 mapper(Address, addresses) 661 mapper(Order, orders) 662 663 sess = create_session() 664 665 q = sess.query(User).options( 666 load_only("name").defaultload("addresses").load_only("id", "email_address"), 667 defaultload("orders").load_only("id") 668 ) 669 670 # hmmmm joinedload seems to be forcing users.id into here... 671 self.assert_compile( 672 q, 673 "SELECT users.id AS users_id, users.name AS users_name, " 674 "addresses_1.id AS addresses_1_id, " 675 "addresses_1.email_address AS addresses_1_email_address, " 676 "orders_1.id AS orders_1_id FROM users " 677 "LEFT OUTER JOIN addresses AS addresses_1 " 678 "ON users.id = addresses_1.user_id " 679 "LEFT OUTER JOIN orders AS orders_1 ON users.id = orders_1.user_id" 680 ) 681 682 683class InheritanceTest(_Polymorphic): 684 __dialect__ = 'default' 685 686 def test_load_only_subclass(self): 687 s = Session() 688 q = s.query(Manager).options(load_only("status", "manager_name")) 689 self.assert_compile( 690 q, 691 "SELECT managers.person_id AS managers_person_id, " 692 "people.person_id AS people_person_id, " 693 "people.type AS people_type, " 694 "managers.status AS managers_status, " 695 "managers.manager_name AS managers_manager_name " 696 "FROM people JOIN managers " 697 "ON people.person_id = managers.person_id " 698 "ORDER BY people.person_id" 699 ) 700 701 def test_load_only_subclass_and_superclass(self): 702 s = Session() 703 q = s.query(Boss).options(load_only("status", "manager_name")) 704 self.assert_compile( 705 q, 706 "SELECT managers.person_id AS managers_person_id, " 707 "people.person_id AS people_person_id, " 708 "people.type AS people_type, " 709 "managers.status AS managers_status, " 710 "managers.manager_name AS managers_manager_name " 711 "FROM people JOIN managers " 712 "ON people.person_id = managers.person_id JOIN boss " 713 "ON managers.person_id = boss.boss_id ORDER BY people.person_id" 714 ) 715 716 def test_load_only_alias_subclass(self): 717 s = Session() 718 m1 = aliased(Manager, flat=True) 719 q = s.query(m1).options(load_only("status", "manager_name")) 720 self.assert_compile( 721 q, 722 "SELECT managers_1.person_id AS managers_1_person_id, " 723 "people_1.person_id AS people_1_person_id, " 724 "people_1.type AS people_1_type, " 725 "managers_1.status AS managers_1_status, " 726 "managers_1.manager_name AS managers_1_manager_name " 727 "FROM people AS people_1 JOIN managers AS " 728 "managers_1 ON people_1.person_id = managers_1.person_id " 729 "ORDER BY people_1.person_id" 730 ) 731 732 def test_load_only_subclass_from_relationship_polymorphic(self): 733 s = Session() 734 wp = with_polymorphic(Person, [Manager], flat=True) 735 q = s.query(Company).join(Company.employees.of_type(wp)).options( 736 contains_eager(Company.employees.of_type(wp)). 737 load_only(wp.Manager.status, wp.Manager.manager_name) 738 ) 739 self.assert_compile( 740 q, 741 "SELECT people_1.person_id AS people_1_person_id, " 742 "people_1.type AS people_1_type, " 743 "managers_1.person_id AS managers_1_person_id, " 744 "managers_1.status AS managers_1_status, " 745 "managers_1.manager_name AS managers_1_manager_name, " 746 "companies.company_id AS companies_company_id, " 747 "companies.name AS companies_name " 748 "FROM companies JOIN (people AS people_1 LEFT OUTER JOIN " 749 "managers AS managers_1 ON people_1.person_id = " 750 "managers_1.person_id) ON companies.company_id = " 751 "people_1.company_id" 752 ) 753 754 def test_load_only_subclass_from_relationship(self): 755 s = Session() 756 from sqlalchemy import inspect 757 inspect(Company).add_property("managers", relationship(Manager)) 758 q = s.query(Company).join(Company.managers).options( 759 contains_eager(Company.managers). 760 load_only("status", "manager_name") 761 ) 762 self.assert_compile( 763 q, 764 "SELECT companies.company_id AS companies_company_id, " 765 "companies.name AS companies_name, " 766 "managers.person_id AS managers_person_id, " 767 "people.person_id AS people_person_id, " 768 "people.type AS people_type, " 769 "managers.status AS managers_status, " 770 "managers.manager_name AS managers_manager_name " 771 "FROM companies JOIN (people JOIN managers ON people.person_id = " 772 "managers.person_id) ON companies.company_id = people.company_id" 773 ) 774 775 776 def test_defer_on_wildcard_subclass(self): 777 # pretty much the same as load_only except doesn't 778 # exclude the primary key 779 780 s = Session() 781 q = s.query(Manager).options( 782 defer(".*"), undefer("status")) 783 self.assert_compile( 784 q, 785 "SELECT managers.status AS managers_status " 786 "FROM people JOIN managers ON " 787 "people.person_id = managers.person_id ORDER BY people.person_id" 788 ) 789 790 def test_defer_super_name_on_subclass(self): 791 s = Session() 792 q = s.query(Manager).options(defer("name")) 793 self.assert_compile( 794 q, 795 "SELECT managers.person_id AS managers_person_id, " 796 "people.person_id AS people_person_id, " 797 "people.company_id AS people_company_id, " 798 "people.type AS people_type, managers.status AS managers_status, " 799 "managers.manager_name AS managers_manager_name " 800 "FROM people JOIN managers " 801 "ON people.person_id = managers.person_id " 802 "ORDER BY people.person_id" 803 ) 804 805 806 807 808