1"""Attribute/instance expiration, deferral of attributes, etc.""" 2 3import sqlalchemy as sa 4from sqlalchemy import exc as sa_exc 5from sqlalchemy import FetchedValue 6from sqlalchemy import ForeignKey 7from sqlalchemy import Integer 8from sqlalchemy import String 9from sqlalchemy import testing 10from sqlalchemy.orm import attributes 11from sqlalchemy.orm import create_session 12from sqlalchemy.orm import defer 13from sqlalchemy.orm import deferred 14from sqlalchemy.orm import exc as orm_exc 15from sqlalchemy.orm import lazyload 16from sqlalchemy.orm import make_transient_to_detached 17from sqlalchemy.orm import mapper 18from sqlalchemy.orm import relationship 19from sqlalchemy.orm import Session 20from sqlalchemy.orm import strategies 21from sqlalchemy.orm import undefer 22from sqlalchemy.sql import select 23from sqlalchemy.testing import assert_raises 24from sqlalchemy.testing import assert_raises_message 25from sqlalchemy.testing import eq_ 26from sqlalchemy.testing import fixtures 27from sqlalchemy.testing.schema import Column 28from sqlalchemy.testing.schema import Table 29from sqlalchemy.testing.util import gc_collect 30from test.orm import _fixtures 31 32 33class ExpireTest(_fixtures.FixtureTest): 34 def test_expire(self): 35 users, Address, addresses, User = ( 36 self.tables.users, 37 self.classes.Address, 38 self.tables.addresses, 39 self.classes.User, 40 ) 41 42 mapper( 43 User, 44 users, 45 properties={"addresses": relationship(Address, backref="user")}, 46 ) 47 mapper(Address, addresses) 48 49 sess = create_session() 50 u = sess.query(User).get(7) 51 assert len(u.addresses) == 1 52 u.name = "foo" 53 del u.addresses[0] 54 sess.expire(u) 55 56 assert "name" not in u.__dict__ 57 58 def go(): 59 assert u.name == "jack" 60 61 self.assert_sql_count(testing.db, go, 1) 62 assert "name" in u.__dict__ 63 64 u.name = "foo" 65 sess.flush() 66 # change the value in the DB 67 users.update(users.c.id == 7, values=dict(name="jack")).execute() 68 sess.expire(u) 69 # object isn't refreshed yet, using dict to bypass trigger 70 assert u.__dict__.get("name") != "jack" 71 assert "name" in attributes.instance_state(u).expired_attributes 72 73 sess.query(User).all() 74 # test that it refreshed 75 assert u.__dict__["name"] == "jack" 76 assert "name" not in attributes.instance_state(u).expired_attributes 77 78 def go(): 79 assert u.name == "jack" 80 81 self.assert_sql_count(testing.db, go, 0) 82 83 def test_persistence_check(self): 84 users, User = self.tables.users, self.classes.User 85 86 mapper(User, users) 87 s = create_session() 88 u = s.query(User).get(7) 89 s.expunge_all() 90 91 assert_raises_message( 92 sa_exc.InvalidRequestError, 93 r"is not persistent within this Session", 94 s.expire, 95 u, 96 ) 97 98 def test_get_refreshes(self): 99 users, User = self.tables.users, self.classes.User 100 101 mapper(User, users) 102 s = create_session(autocommit=False) 103 u = s.query(User).get(10) 104 s.expire_all() 105 106 def go(): 107 s.query(User).get(10) # get() refreshes 108 109 self.assert_sql_count(testing.db, go, 1) 110 111 def go(): 112 eq_(u.name, "chuck") # attributes unexpired 113 114 self.assert_sql_count(testing.db, go, 0) 115 116 def go(): 117 s.query(User).get(10) # expire flag reset, so not expired 118 119 self.assert_sql_count(testing.db, go, 0) 120 121 def test_get_on_deleted_expunges(self): 122 users, User = self.tables.users, self.classes.User 123 124 mapper(User, users) 125 s = create_session(autocommit=False) 126 u = s.query(User).get(10) 127 128 s.expire_all() 129 s.execute(users.delete().where(User.id == 10)) 130 131 # object is gone, get() returns None, removes u from session 132 assert u in s 133 assert s.query(User).get(10) is None 134 assert u not in s # and expunges 135 136 def test_refresh_on_deleted_raises(self): 137 users, User = self.tables.users, self.classes.User 138 139 mapper(User, users) 140 s = create_session(autocommit=False) 141 u = s.query(User).get(10) 142 s.expire_all() 143 144 s.expire_all() 145 s.execute(users.delete().where(User.id == 10)) 146 147 # raises ObjectDeletedError 148 assert_raises_message( 149 sa.orm.exc.ObjectDeletedError, 150 "Instance '<User at .*?>' has been " 151 "deleted, or its row is otherwise not present.", 152 getattr, 153 u, 154 "name", 155 ) 156 157 def test_rollback_undoes_expunge_from_deleted(self): 158 users, User = self.tables.users, self.classes.User 159 160 mapper(User, users) 161 s = create_session(autocommit=False) 162 u = s.query(User).get(10) 163 s.expire_all() 164 s.execute(users.delete().where(User.id == 10)) 165 166 # do a get()/remove u from session 167 assert s.query(User).get(10) is None 168 assert u not in s 169 170 s.rollback() 171 172 assert u in s 173 # but now its back, rollback has occurred, the 174 # _remove_newly_deleted is reverted 175 eq_(u.name, "chuck") 176 177 def test_deferred(self): 178 """test that unloaded, deferred attributes aren't included in the 179 expiry list.""" 180 181 Order, orders = self.classes.Order, self.tables.orders 182 183 mapper( 184 Order, 185 orders, 186 properties={"description": deferred(orders.c.description)}, 187 ) 188 189 s = create_session() 190 o1 = s.query(Order).first() 191 assert "description" not in o1.__dict__ 192 s.expire(o1) 193 assert o1.isopen is not None 194 assert "description" not in o1.__dict__ 195 assert o1.description 196 197 def test_deferred_notfound(self): 198 users, User = self.tables.users, self.classes.User 199 200 mapper(User, users, properties={"name": deferred(users.c.name)}) 201 s = create_session(autocommit=False) 202 u = s.query(User).get(10) 203 204 assert "name" not in u.__dict__ 205 s.execute(users.delete().where(User.id == 10)) 206 assert_raises_message( 207 sa.orm.exc.ObjectDeletedError, 208 "Instance '<User at .*?>' has been " 209 "deleted, or its row is otherwise not present.", 210 getattr, 211 u, 212 "name", 213 ) 214 215 def test_lazyload_autoflushes(self): 216 users, Address, addresses, User = ( 217 self.tables.users, 218 self.classes.Address, 219 self.tables.addresses, 220 self.classes.User, 221 ) 222 223 mapper( 224 User, 225 users, 226 properties={ 227 "addresses": relationship( 228 Address, order_by=addresses.c.email_address 229 ) 230 }, 231 ) 232 mapper(Address, addresses) 233 s = create_session(autoflush=True, autocommit=False) 234 u = s.query(User).get(8) 235 adlist = u.addresses 236 eq_( 237 adlist, 238 [ 239 Address(email_address="ed@bettyboop.com"), 240 Address(email_address="ed@lala.com"), 241 Address(email_address="ed@wood.com"), 242 ], 243 ) 244 a1 = u.addresses[2] 245 a1.email_address = "aaaaa" 246 s.expire(u, ["addresses"]) 247 eq_( 248 u.addresses, 249 [ 250 Address(email_address="aaaaa"), 251 Address(email_address="ed@bettyboop.com"), 252 Address(email_address="ed@lala.com"), 253 ], 254 ) 255 256 def test_refresh_collection_exception(self): 257 """test graceful failure for currently unsupported 258 immediate refresh of a collection""" 259 260 users, Address, addresses, User = ( 261 self.tables.users, 262 self.classes.Address, 263 self.tables.addresses, 264 self.classes.User, 265 ) 266 267 mapper( 268 User, 269 users, 270 properties={ 271 "addresses": relationship( 272 Address, order_by=addresses.c.email_address 273 ) 274 }, 275 ) 276 mapper(Address, addresses) 277 s = create_session(autoflush=True, autocommit=False) 278 u = s.query(User).get(8) 279 assert_raises_message( 280 sa_exc.InvalidRequestError, 281 "properties specified for refresh", 282 s.refresh, 283 u, 284 ["addresses"], 285 ) 286 287 # in contrast to a regular query with no columns 288 assert_raises_message( 289 sa_exc.InvalidRequestError, 290 "no columns with which to SELECT", 291 s.query().all, 292 ) 293 294 def test_refresh_cancels_expire(self): 295 users, User = self.tables.users, self.classes.User 296 297 mapper(User, users) 298 s = create_session() 299 u = s.query(User).get(7) 300 s.expire(u) 301 s.refresh(u) 302 303 def go(): 304 u = s.query(User).get(7) 305 eq_(u.name, "jack") 306 307 self.assert_sql_count(testing.db, go, 0) 308 309 def test_expire_doesntload_on_set(self): 310 User, users = self.classes.User, self.tables.users 311 312 mapper(User, users) 313 314 sess = create_session() 315 u = sess.query(User).get(7) 316 317 sess.expire(u, attribute_names=["name"]) 318 319 def go(): 320 u.name = "somenewname" 321 322 self.assert_sql_count(testing.db, go, 0) 323 sess.flush() 324 sess.expunge_all() 325 assert sess.query(User).get(7).name == "somenewname" 326 327 def test_no_session(self): 328 users, User = self.tables.users, self.classes.User 329 330 mapper(User, users) 331 sess = create_session() 332 u = sess.query(User).get(7) 333 334 sess.expire(u, attribute_names=["name"]) 335 sess.expunge(u) 336 assert_raises(orm_exc.DetachedInstanceError, getattr, u, "name") 337 338 def test_pending_raises(self): 339 users, User = self.tables.users, self.classes.User 340 341 # this was the opposite in 0.4, but the reasoning there seemed off. 342 # expiring a pending instance makes no sense, so should raise 343 mapper(User, users) 344 sess = create_session() 345 u = User(id=15) 346 sess.add(u) 347 assert_raises(sa_exc.InvalidRequestError, sess.expire, u, ["name"]) 348 349 def test_no_instance_key(self): 350 User, users = self.classes.User, self.tables.users 351 352 # this tests an artificial condition such that 353 # an instance is pending, but has expired attributes. this 354 # is actually part of a larger behavior when postfetch needs to 355 # occur during a flush() on an instance that was just inserted 356 mapper(User, users) 357 sess = create_session() 358 u = sess.query(User).get(7) 359 360 sess.expire(u, attribute_names=["name"]) 361 sess.expunge(u) 362 attributes.instance_state(u).key = None 363 assert "name" not in u.__dict__ 364 sess.add(u) 365 assert u.name == "jack" 366 367 def test_no_instance_key_no_pk(self): 368 users, User = self.tables.users, self.classes.User 369 370 # same as test_no_instance_key, but the PK columns 371 # are absent. ensure an error is raised. 372 mapper(User, users) 373 sess = create_session() 374 u = sess.query(User).get(7) 375 376 sess.expire(u, attribute_names=["name", "id"]) 377 sess.expunge(u) 378 attributes.instance_state(u).key = None 379 assert "name" not in u.__dict__ 380 sess.add(u) 381 assert_raises(sa_exc.InvalidRequestError, getattr, u, "name") 382 383 def test_expire_preserves_changes(self): 384 """test that the expire load operation doesn't revert post-expire 385 changes""" 386 387 Order, orders = self.classes.Order, self.tables.orders 388 389 mapper(Order, orders) 390 sess = create_session() 391 o = sess.query(Order).get(3) 392 sess.expire(o) 393 394 o.description = "order 3 modified" 395 396 def go(): 397 assert o.isopen == 1 398 399 self.assert_sql_count(testing.db, go, 1) 400 assert o.description == "order 3 modified" 401 402 del o.description 403 assert "description" not in o.__dict__ 404 sess.expire(o, ["isopen"]) 405 sess.query(Order).all() 406 assert o.isopen == 1 407 assert "description" not in o.__dict__ 408 409 assert o.description is None 410 411 o.isopen = 15 412 sess.expire(o, ["isopen", "description"]) 413 o.description = "some new description" 414 sess.query(Order).all() 415 assert o.isopen == 1 416 assert o.description == "some new description" 417 418 sess.expire(o, ["isopen", "description"]) 419 sess.query(Order).all() 420 del o.isopen 421 422 def go(): 423 assert o.isopen is None 424 425 self.assert_sql_count(testing.db, go, 0) 426 427 o.isopen = 14 428 sess.expire(o) 429 o.description = "another new description" 430 sess.query(Order).all() 431 assert o.isopen == 1 432 assert o.description == "another new description" 433 434 def test_expire_committed(self): 435 """test that the committed state of the attribute receives the most 436 recent DB data""" 437 438 orders, Order = self.tables.orders, self.classes.Order 439 440 mapper(Order, orders) 441 442 sess = create_session() 443 o = sess.query(Order).get(3) 444 sess.expire(o) 445 446 orders.update().execute(description="order 3 modified") 447 assert o.isopen == 1 448 assert ( 449 attributes.instance_state(o).dict["description"] 450 == "order 3 modified" 451 ) 452 453 def go(): 454 sess.flush() 455 456 self.assert_sql_count(testing.db, go, 0) 457 458 def test_expire_cascade(self): 459 users, Address, addresses, User = ( 460 self.tables.users, 461 self.classes.Address, 462 self.tables.addresses, 463 self.classes.User, 464 ) 465 466 mapper( 467 User, 468 users, 469 properties={ 470 "addresses": relationship( 471 Address, cascade="all, refresh-expire" 472 ) 473 }, 474 ) 475 mapper(Address, addresses) 476 s = create_session() 477 u = s.query(User).get(8) 478 assert u.addresses[0].email_address == "ed@wood.com" 479 480 u.addresses[0].email_address = "someotheraddress" 481 s.expire(u) 482 assert u.addresses[0].email_address == "ed@wood.com" 483 484 def test_refresh_cascade(self): 485 users, Address, addresses, User = ( 486 self.tables.users, 487 self.classes.Address, 488 self.tables.addresses, 489 self.classes.User, 490 ) 491 492 mapper( 493 User, 494 users, 495 properties={ 496 "addresses": relationship( 497 Address, cascade="all, refresh-expire" 498 ) 499 }, 500 ) 501 mapper(Address, addresses) 502 s = create_session() 503 u = s.query(User).get(8) 504 assert u.addresses[0].email_address == "ed@wood.com" 505 506 u.addresses[0].email_address = "someotheraddress" 507 s.refresh(u) 508 assert u.addresses[0].email_address == "ed@wood.com" 509 510 def test_expire_cascade_pending_orphan(self): 511 cascade = "save-update, refresh-expire, delete, delete-orphan" 512 self._test_cascade_to_pending(cascade, True) 513 514 def test_refresh_cascade_pending_orphan(self): 515 cascade = "save-update, refresh-expire, delete, delete-orphan" 516 self._test_cascade_to_pending(cascade, False) 517 518 def test_expire_cascade_pending(self): 519 cascade = "save-update, refresh-expire" 520 self._test_cascade_to_pending(cascade, True) 521 522 def test_refresh_cascade_pending(self): 523 cascade = "save-update, refresh-expire" 524 self._test_cascade_to_pending(cascade, False) 525 526 def _test_cascade_to_pending(self, cascade, expire_or_refresh): 527 users, Address, addresses, User = ( 528 self.tables.users, 529 self.classes.Address, 530 self.tables.addresses, 531 self.classes.User, 532 ) 533 534 mapper( 535 User, 536 users, 537 properties={"addresses": relationship(Address, cascade=cascade)}, 538 ) 539 mapper(Address, addresses) 540 s = create_session() 541 542 u = s.query(User).get(8) 543 a = Address(id=12, email_address="foobar") 544 545 u.addresses.append(a) 546 if expire_or_refresh: 547 s.expire(u) 548 else: 549 s.refresh(u) 550 if "delete-orphan" in cascade: 551 assert a not in s 552 else: 553 assert a in s 554 555 assert a not in u.addresses 556 s.flush() 557 558 def test_expired_lazy(self): 559 users, Address, addresses, User = ( 560 self.tables.users, 561 self.classes.Address, 562 self.tables.addresses, 563 self.classes.User, 564 ) 565 566 mapper( 567 User, 568 users, 569 properties={"addresses": relationship(Address, backref="user")}, 570 ) 571 mapper(Address, addresses) 572 573 sess = create_session() 574 u = sess.query(User).get(7) 575 576 sess.expire(u) 577 assert "name" not in u.__dict__ 578 assert "addresses" not in u.__dict__ 579 580 def go(): 581 assert u.addresses[0].email_address == "jack@bean.com" 582 assert u.name == "jack" 583 584 # two loads 585 self.assert_sql_count(testing.db, go, 2) 586 assert "name" in u.__dict__ 587 assert "addresses" in u.__dict__ 588 589 def test_expired_eager(self): 590 users, Address, addresses, User = ( 591 self.tables.users, 592 self.classes.Address, 593 self.tables.addresses, 594 self.classes.User, 595 ) 596 597 mapper( 598 User, 599 users, 600 properties={ 601 "addresses": relationship( 602 Address, backref="user", lazy="joined" 603 ) 604 }, 605 ) 606 mapper(Address, addresses) 607 608 sess = create_session() 609 u = sess.query(User).get(7) 610 611 sess.expire(u) 612 assert "name" not in u.__dict__ 613 assert "addresses" not in u.__dict__ 614 615 def go(): 616 assert u.addresses[0].email_address == "jack@bean.com" 617 assert u.name == "jack" 618 619 # two loads, since relationship() + scalar are 620 # separate right now on per-attribute load 621 self.assert_sql_count(testing.db, go, 2) 622 assert "name" in u.__dict__ 623 assert "addresses" in u.__dict__ 624 625 sess.expire(u, ["name", "addresses"]) 626 assert "name" not in u.__dict__ 627 assert "addresses" not in u.__dict__ 628 629 def go(): 630 sess.query(User).filter_by(id=7).one() 631 assert u.addresses[0].email_address == "jack@bean.com" 632 assert u.name == "jack" 633 634 # one load, since relationship() + scalar are 635 # together when eager load used with Query 636 self.assert_sql_count(testing.db, go, 1) 637 638 def test_relationship_changes_preserved(self): 639 users, Address, addresses, User = ( 640 self.tables.users, 641 self.classes.Address, 642 self.tables.addresses, 643 self.classes.User, 644 ) 645 646 mapper( 647 User, 648 users, 649 properties={ 650 "addresses": relationship( 651 Address, backref="user", lazy="joined" 652 ) 653 }, 654 ) 655 mapper(Address, addresses) 656 sess = create_session() 657 u = sess.query(User).get(8) 658 sess.expire(u, ["name", "addresses"]) 659 u.addresses 660 assert "name" not in u.__dict__ 661 del u.addresses[1] 662 u.name 663 assert "name" in u.__dict__ 664 assert len(u.addresses) == 2 665 666 def test_joinedload_props_dontload(self): 667 users, Address, addresses, User = ( 668 self.tables.users, 669 self.classes.Address, 670 self.tables.addresses, 671 self.classes.User, 672 ) 673 674 # relationships currently have to load separately from scalar instances 675 # the use case is: expire "addresses". then access it. lazy load 676 # fires off to load "addresses", but needs foreign key or primary key 677 # attributes in order to lazy load; hits those attributes, such as 678 # below it hits "u.id". "u.id" triggers full unexpire operation, 679 # joinedloads addresses since lazy='joined'. this is all within lazy 680 # load which fires unconditionally; so an unnecessary joinedload (or 681 # lazyload) was issued. would prefer not to complicate lazyloading to 682 # "figure out" that the operation should be aborted right now. 683 684 mapper( 685 User, 686 users, 687 properties={ 688 "addresses": relationship( 689 Address, backref="user", lazy="joined" 690 ) 691 }, 692 ) 693 mapper(Address, addresses) 694 sess = create_session() 695 u = sess.query(User).get(8) 696 sess.expire(u) 697 u.id 698 assert "addresses" not in u.__dict__ 699 u.addresses 700 assert "addresses" in u.__dict__ 701 702 def test_expire_synonym(self): 703 User, users = self.classes.User, self.tables.users 704 705 mapper(User, users, properties={"uname": sa.orm.synonym("name")}) 706 707 sess = create_session() 708 u = sess.query(User).get(7) 709 assert "name" in u.__dict__ 710 assert u.uname == u.name 711 712 sess.expire(u) 713 assert "name" not in u.__dict__ 714 715 users.update(users.c.id == 7).execute(name="jack2") 716 assert u.name == "jack2" 717 assert u.uname == "jack2" 718 assert "name" in u.__dict__ 719 720 # this wont work unless we add API hooks through the attr. system to 721 # provide "expire" behavior on a synonym 722 # sess.expire(u, ['uname']) 723 # users.update(users.c.id==7).execute(name='jack3') 724 # assert u.uname == 'jack3' 725 726 def test_partial_expire(self): 727 orders, Order = self.tables.orders, self.classes.Order 728 729 mapper(Order, orders) 730 731 sess = create_session() 732 o = sess.query(Order).get(3) 733 734 sess.expire(o, attribute_names=["description"]) 735 assert "id" in o.__dict__ 736 assert "description" not in o.__dict__ 737 assert attributes.instance_state(o).dict["isopen"] == 1 738 739 orders.update(orders.c.id == 3).execute(description="order 3 modified") 740 741 def go(): 742 assert o.description == "order 3 modified" 743 744 self.assert_sql_count(testing.db, go, 1) 745 assert ( 746 attributes.instance_state(o).dict["description"] 747 == "order 3 modified" 748 ) 749 750 o.isopen = 5 751 sess.expire(o, attribute_names=["description"]) 752 assert "id" in o.__dict__ 753 assert "description" not in o.__dict__ 754 assert o.__dict__["isopen"] == 5 755 assert attributes.instance_state(o).committed_state["isopen"] == 1 756 757 def go(): 758 assert o.description == "order 3 modified" 759 760 self.assert_sql_count(testing.db, go, 1) 761 assert o.__dict__["isopen"] == 5 762 assert ( 763 attributes.instance_state(o).dict["description"] 764 == "order 3 modified" 765 ) 766 assert attributes.instance_state(o).committed_state["isopen"] == 1 767 768 sess.flush() 769 770 sess.expire(o, attribute_names=["id", "isopen", "description"]) 771 assert "id" not in o.__dict__ 772 assert "isopen" not in o.__dict__ 773 assert "description" not in o.__dict__ 774 775 def go(): 776 assert o.description == "order 3 modified" 777 assert o.id == 3 778 assert o.isopen == 5 779 780 self.assert_sql_count(testing.db, go, 1) 781 782 def test_partial_expire_lazy(self): 783 users, Address, addresses, User = ( 784 self.tables.users, 785 self.classes.Address, 786 self.tables.addresses, 787 self.classes.User, 788 ) 789 790 mapper( 791 User, 792 users, 793 properties={"addresses": relationship(Address, backref="user")}, 794 ) 795 mapper(Address, addresses) 796 797 sess = create_session() 798 u = sess.query(User).get(8) 799 800 sess.expire(u, ["name", "addresses"]) 801 assert "name" not in u.__dict__ 802 assert "addresses" not in u.__dict__ 803 804 # hit the lazy loader. just does the lazy load, 805 # doesn't do the overall refresh 806 def go(): 807 assert u.addresses[0].email_address == "ed@wood.com" 808 809 self.assert_sql_count(testing.db, go, 1) 810 811 assert "name" not in u.__dict__ 812 813 # check that mods to expired lazy-load attributes 814 # only do the lazy load 815 sess.expire(u, ["name", "addresses"]) 816 817 def go(): 818 u.addresses = [Address(id=10, email_address="foo@bar.com")] 819 820 self.assert_sql_count(testing.db, go, 1) 821 822 sess.flush() 823 824 # flush has occurred, and addresses was modified, 825 # so the addresses collection got committed and is 826 # longer expired 827 def go(): 828 assert u.addresses[0].email_address == "foo@bar.com" 829 assert len(u.addresses) == 1 830 831 self.assert_sql_count(testing.db, go, 0) 832 833 # but the name attribute was never loaded and so 834 # still loads 835 def go(): 836 assert u.name == "ed" 837 838 self.assert_sql_count(testing.db, go, 1) 839 840 def test_partial_expire_eager(self): 841 users, Address, addresses, User = ( 842 self.tables.users, 843 self.classes.Address, 844 self.tables.addresses, 845 self.classes.User, 846 ) 847 848 mapper( 849 User, 850 users, 851 properties={ 852 "addresses": relationship( 853 Address, backref="user", lazy="joined" 854 ) 855 }, 856 ) 857 mapper(Address, addresses) 858 859 sess = create_session() 860 u = sess.query(User).get(8) 861 862 sess.expire(u, ["name", "addresses"]) 863 assert "name" not in u.__dict__ 864 assert "addresses" not in u.__dict__ 865 866 def go(): 867 assert u.addresses[0].email_address == "ed@wood.com" 868 869 self.assert_sql_count(testing.db, go, 1) 870 871 # check that mods to expired eager-load attributes 872 # do the refresh 873 sess.expire(u, ["name", "addresses"]) 874 875 def go(): 876 u.addresses = [Address(id=10, email_address="foo@bar.com")] 877 878 self.assert_sql_count(testing.db, go, 1) 879 sess.flush() 880 881 # this should ideally trigger the whole load 882 # but currently it works like the lazy case 883 def go(): 884 assert u.addresses[0].email_address == "foo@bar.com" 885 assert len(u.addresses) == 1 886 887 self.assert_sql_count(testing.db, go, 0) 888 889 def go(): 890 assert u.name == "ed" 891 892 # scalar attributes have their own load 893 self.assert_sql_count(testing.db, go, 1) 894 # ideally, this was already loaded, but we aren't 895 # doing it that way right now 896 # self.assert_sql_count(testing.db, go, 0) 897 898 def test_relationships_load_on_query(self): 899 users, Address, addresses, User = ( 900 self.tables.users, 901 self.classes.Address, 902 self.tables.addresses, 903 self.classes.User, 904 ) 905 906 mapper( 907 User, 908 users, 909 properties={"addresses": relationship(Address, backref="user")}, 910 ) 911 mapper(Address, addresses) 912 913 sess = create_session() 914 u = sess.query(User).get(8) 915 assert "name" in u.__dict__ 916 u.addresses 917 assert "addresses" in u.__dict__ 918 919 sess.expire(u, ["name", "addresses"]) 920 assert "name" not in u.__dict__ 921 assert "addresses" not in u.__dict__ 922 ( 923 sess.query(User) 924 .options(sa.orm.joinedload("addresses")) 925 .filter_by(id=8) 926 .all() 927 ) 928 assert "name" in u.__dict__ 929 assert "addresses" in u.__dict__ 930 931 def test_partial_expire_deferred(self): 932 orders, Order = self.tables.orders, self.classes.Order 933 934 mapper( 935 Order, 936 orders, 937 properties={"description": sa.orm.deferred(orders.c.description)}, 938 ) 939 940 sess = create_session() 941 o = sess.query(Order).get(3) 942 sess.expire(o, ["description", "isopen"]) 943 assert "isopen" not in o.__dict__ 944 assert "description" not in o.__dict__ 945 946 # test that expired attribute access refreshes 947 # the deferred 948 def go(): 949 assert o.isopen == 1 950 assert o.description == "order 3" 951 952 self.assert_sql_count(testing.db, go, 1) 953 954 sess.expire(o, ["description", "isopen"]) 955 assert "isopen" not in o.__dict__ 956 assert "description" not in o.__dict__ 957 # test that the deferred attribute triggers the full 958 # reload 959 960 def go(): 961 assert o.description == "order 3" 962 assert o.isopen == 1 963 964 self.assert_sql_count(testing.db, go, 1) 965 966 sa.orm.clear_mappers() 967 968 mapper(Order, orders) 969 sess.expunge_all() 970 971 # same tests, using deferred at the options level 972 o = sess.query(Order).options(sa.orm.defer("description")).get(3) 973 974 assert "description" not in o.__dict__ 975 976 # sanity check 977 def go(): 978 assert o.description == "order 3" 979 980 self.assert_sql_count(testing.db, go, 1) 981 982 assert "description" in o.__dict__ 983 assert "isopen" in o.__dict__ 984 sess.expire(o, ["description", "isopen"]) 985 assert "isopen" not in o.__dict__ 986 assert "description" not in o.__dict__ 987 988 # test that expired attribute access refreshes 989 # the deferred 990 def go(): 991 assert o.isopen == 1 992 assert o.description == "order 3" 993 994 self.assert_sql_count(testing.db, go, 1) 995 sess.expire(o, ["description", "isopen"]) 996 997 assert "isopen" not in o.__dict__ 998 assert "description" not in o.__dict__ 999 # test that the deferred attribute triggers the full 1000 # reload 1001 1002 def go(): 1003 assert o.description == "order 3" 1004 assert o.isopen == 1 1005 1006 self.assert_sql_count(testing.db, go, 1) 1007 1008 def test_joinedload_query_refreshes(self): 1009 users, Address, addresses, User = ( 1010 self.tables.users, 1011 self.classes.Address, 1012 self.tables.addresses, 1013 self.classes.User, 1014 ) 1015 1016 mapper( 1017 User, 1018 users, 1019 properties={ 1020 "addresses": relationship( 1021 Address, backref="user", lazy="joined" 1022 ) 1023 }, 1024 ) 1025 mapper(Address, addresses) 1026 1027 sess = create_session() 1028 u = sess.query(User).get(8) 1029 assert len(u.addresses) == 3 1030 sess.expire(u) 1031 assert "addresses" not in u.__dict__ 1032 sess.query(User).filter_by(id=8).all() 1033 assert "addresses" in u.__dict__ 1034 assert len(u.addresses) == 3 1035 1036 @testing.requires.predictable_gc 1037 def test_expire_all(self): 1038 users, Address, addresses, User = ( 1039 self.tables.users, 1040 self.classes.Address, 1041 self.tables.addresses, 1042 self.classes.User, 1043 ) 1044 1045 mapper( 1046 User, 1047 users, 1048 properties={ 1049 "addresses": relationship( 1050 Address, 1051 backref="user", 1052 lazy="joined", 1053 order_by=addresses.c.id, 1054 ) 1055 }, 1056 ) 1057 mapper(Address, addresses) 1058 1059 sess = create_session() 1060 userlist = sess.query(User).order_by(User.id).all() 1061 eq_(self.static.user_address_result, userlist) 1062 eq_(len(list(sess)), 9) 1063 sess.expire_all() 1064 gc_collect() 1065 eq_(len(list(sess)), 4) # since addresses were gc'ed 1066 1067 userlist = sess.query(User).order_by(User.id).all() 1068 eq_(self.static.user_address_result, userlist) 1069 eq_(len(list(sess)), 9) 1070 1071 def test_state_change_col_to_deferred(self): 1072 """Behavioral test to verify the current activity of loader 1073 callables""" 1074 1075 users, User = self.tables.users, self.classes.User 1076 1077 mapper(User, users) 1078 1079 sess = create_session() 1080 1081 # deferred attribute option, gets the LoadDeferredColumns 1082 # callable 1083 u1 = sess.query(User).options(defer(User.name)).first() 1084 assert isinstance( 1085 attributes.instance_state(u1).callables["name"], 1086 strategies.LoadDeferredColumns, 1087 ) 1088 1089 # expire the attr, it gets the InstanceState callable 1090 sess.expire(u1, ["name"]) 1091 assert "name" in attributes.instance_state(u1).expired_attributes 1092 assert "name" not in attributes.instance_state(u1).callables 1093 1094 # load it, callable is gone 1095 u1.name 1096 assert "name" not in attributes.instance_state(u1).expired_attributes 1097 assert "name" not in attributes.instance_state(u1).callables 1098 1099 # same for expire all 1100 sess.expunge_all() 1101 u1 = sess.query(User).options(defer(User.name)).first() 1102 sess.expire(u1) 1103 assert "name" in attributes.instance_state(u1).expired_attributes 1104 assert "name" not in attributes.instance_state(u1).callables 1105 1106 # load over it. everything normal. 1107 sess.query(User).first() 1108 assert "name" not in attributes.instance_state(u1).expired_attributes 1109 assert "name" not in attributes.instance_state(u1).callables 1110 1111 sess.expunge_all() 1112 u1 = sess.query(User).first() 1113 # for non present, still expires the same way 1114 del u1.name 1115 sess.expire(u1) 1116 assert "name" in attributes.instance_state(u1).expired_attributes 1117 assert "name" not in attributes.instance_state(u1).callables 1118 1119 def test_state_deferred_to_col(self): 1120 """Behavioral test to verify the current activity of loader 1121 callables""" 1122 1123 users, User = self.tables.users, self.classes.User 1124 1125 mapper(User, users, properties={"name": deferred(users.c.name)}) 1126 1127 sess = create_session() 1128 u1 = sess.query(User).options(undefer(User.name)).first() 1129 assert "name" not in attributes.instance_state(u1).callables 1130 1131 # mass expire, the attribute was loaded, 1132 # the attribute gets the callable 1133 sess.expire(u1) 1134 assert "name" in attributes.instance_state(u1).expired_attributes 1135 assert "name" not in attributes.instance_state(u1).callables 1136 1137 # load it 1138 u1.name 1139 assert "name" not in attributes.instance_state(u1).expired_attributes 1140 assert "name" not in attributes.instance_state(u1).callables 1141 1142 # mass expire, attribute was loaded but then deleted, 1143 # the callable goes away - the state wants to flip 1144 # it back to its "deferred" loader. 1145 sess.expunge_all() 1146 u1 = sess.query(User).options(undefer(User.name)).first() 1147 del u1.name 1148 sess.expire(u1) 1149 assert "name" not in attributes.instance_state(u1).expired_attributes 1150 assert "name" not in attributes.instance_state(u1).callables 1151 1152 # single attribute expire, the attribute gets the callable 1153 sess.expunge_all() 1154 u1 = sess.query(User).options(undefer(User.name)).first() 1155 sess.expire(u1, ["name"]) 1156 assert "name" in attributes.instance_state(u1).expired_attributes 1157 assert "name" not in attributes.instance_state(u1).callables 1158 1159 def test_state_noload_to_lazy(self): 1160 """Behavioral test to verify the current activity of loader 1161 callables""" 1162 1163 users, Address, addresses, User = ( 1164 self.tables.users, 1165 self.classes.Address, 1166 self.tables.addresses, 1167 self.classes.User, 1168 ) 1169 1170 mapper( 1171 User, 1172 users, 1173 properties={"addresses": relationship(Address, lazy="noload")}, 1174 ) 1175 mapper(Address, addresses) 1176 1177 sess = create_session() 1178 u1 = sess.query(User).options(lazyload(User.addresses)).first() 1179 assert isinstance( 1180 attributes.instance_state(u1).callables["addresses"], 1181 strategies.LoadLazyAttribute, 1182 ) 1183 # expire, it stays 1184 sess.expire(u1) 1185 assert ( 1186 "addresses" not in attributes.instance_state(u1).expired_attributes 1187 ) 1188 assert isinstance( 1189 attributes.instance_state(u1).callables["addresses"], 1190 strategies.LoadLazyAttribute, 1191 ) 1192 1193 # load over it. callable goes away. 1194 sess.query(User).first() 1195 assert ( 1196 "addresses" not in attributes.instance_state(u1).expired_attributes 1197 ) 1198 assert "addresses" not in attributes.instance_state(u1).callables 1199 1200 sess.expunge_all() 1201 u1 = sess.query(User).options(lazyload(User.addresses)).first() 1202 sess.expire(u1, ["addresses"]) 1203 assert ( 1204 "addresses" not in attributes.instance_state(u1).expired_attributes 1205 ) 1206 assert isinstance( 1207 attributes.instance_state(u1).callables["addresses"], 1208 strategies.LoadLazyAttribute, 1209 ) 1210 1211 # load the attr, goes away 1212 u1.addresses 1213 assert ( 1214 "addresses" not in attributes.instance_state(u1).expired_attributes 1215 ) 1216 assert "addresses" not in attributes.instance_state(u1).callables 1217 1218 def test_deferred_expire_w_transient_to_detached(self): 1219 orders, Order = self.tables.orders, self.classes.Order 1220 mapper( 1221 Order, 1222 orders, 1223 properties={"description": deferred(orders.c.description)}, 1224 ) 1225 1226 s = Session() 1227 item = Order(id=1) 1228 1229 make_transient_to_detached(item) 1230 s.add(item) 1231 item.isopen 1232 assert "description" not in item.__dict__ 1233 1234 def test_deferred_expire_normally(self): 1235 orders, Order = self.tables.orders, self.classes.Order 1236 mapper( 1237 Order, 1238 orders, 1239 properties={"description": deferred(orders.c.description)}, 1240 ) 1241 1242 s = Session() 1243 1244 item = s.query(Order).first() 1245 s.expire(item) 1246 item.isopen 1247 assert "description" not in item.__dict__ 1248 1249 def test_deferred_expire_explicit_attrs(self): 1250 orders, Order = self.tables.orders, self.classes.Order 1251 mapper( 1252 Order, 1253 orders, 1254 properties={"description": deferred(orders.c.description)}, 1255 ) 1256 1257 s = Session() 1258 1259 item = s.query(Order).first() 1260 s.expire(item, ["isopen", "description"]) 1261 item.isopen 1262 assert "description" in item.__dict__ 1263 1264 1265class PolymorphicExpireTest(fixtures.MappedTest): 1266 run_inserts = "once" 1267 run_deletes = None 1268 1269 @classmethod 1270 def define_tables(cls, metadata): 1271 Table( 1272 "people", 1273 metadata, 1274 Column( 1275 "person_id", 1276 Integer, 1277 primary_key=True, 1278 test_needs_autoincrement=True, 1279 ), 1280 Column("name", String(50)), 1281 Column("type", String(30)), 1282 ) 1283 1284 Table( 1285 "engineers", 1286 metadata, 1287 Column( 1288 "person_id", 1289 Integer, 1290 ForeignKey("people.person_id"), 1291 primary_key=True, 1292 ), 1293 Column("status", String(30)), 1294 ) 1295 1296 @classmethod 1297 def setup_classes(cls): 1298 class Person(cls.Basic): 1299 pass 1300 1301 class Engineer(Person): 1302 pass 1303 1304 @classmethod 1305 def insert_data(cls, connection): 1306 people, engineers = cls.tables.people, cls.tables.engineers 1307 1308 connection.execute( 1309 people.insert(), 1310 {"person_id": 1, "name": "person1", "type": "person"}, 1311 {"person_id": 2, "name": "engineer1", "type": "engineer"}, 1312 {"person_id": 3, "name": "engineer2", "type": "engineer"}, 1313 ) 1314 connection.execute( 1315 engineers.insert(), 1316 {"person_id": 2, "status": "new engineer"}, 1317 {"person_id": 3, "status": "old engineer"}, 1318 ) 1319 1320 @classmethod 1321 def setup_mappers(cls): 1322 Person, people, engineers, Engineer = ( 1323 cls.classes.Person, 1324 cls.tables.people, 1325 cls.tables.engineers, 1326 cls.classes.Engineer, 1327 ) 1328 1329 mapper( 1330 Person, 1331 people, 1332 polymorphic_on=people.c.type, 1333 polymorphic_identity="person", 1334 ) 1335 mapper( 1336 Engineer, 1337 engineers, 1338 inherits=Person, 1339 polymorphic_identity="engineer", 1340 ) 1341 1342 def test_poly_deferred(self): 1343 Person, people, Engineer = ( 1344 self.classes.Person, 1345 self.tables.people, 1346 self.classes.Engineer, 1347 ) 1348 1349 sess = create_session() 1350 [p1, e1, e2] = sess.query(Person).order_by(people.c.person_id).all() 1351 1352 sess.expire(p1) 1353 sess.expire(e1, ["status"]) 1354 sess.expire(e2) 1355 1356 for p in [p1, e2]: 1357 assert "name" not in p.__dict__ 1358 1359 assert "name" in e1.__dict__ 1360 assert "status" not in e2.__dict__ 1361 assert "status" not in e1.__dict__ 1362 1363 e1.name = "new engineer name" 1364 1365 def go(): 1366 sess.query(Person).all() 1367 1368 self.assert_sql_count(testing.db, go, 1) 1369 1370 for p in [p1, e1, e2]: 1371 assert "name" in p.__dict__ 1372 1373 assert "status" not in e2.__dict__ 1374 assert "status" not in e1.__dict__ 1375 1376 def go(): 1377 assert e1.name == "new engineer name" 1378 assert e2.name == "engineer2" 1379 assert e1.status == "new engineer" 1380 assert e2.status == "old engineer" 1381 1382 self.assert_sql_count(testing.db, go, 2) 1383 eq_( 1384 Engineer.name.get_history(e1), 1385 (["new engineer name"], (), ["engineer1"]), 1386 ) 1387 1388 def test_no_instance_key(self): 1389 Engineer = self.classes.Engineer 1390 1391 sess = create_session() 1392 e1 = sess.query(Engineer).get(2) 1393 1394 sess.expire(e1, attribute_names=["name"]) 1395 sess.expunge(e1) 1396 attributes.instance_state(e1).key = None 1397 assert "name" not in e1.__dict__ 1398 sess.add(e1) 1399 assert e1.name == "engineer1" 1400 1401 def test_no_instance_key_pk_absent(self): 1402 Engineer = self.classes.Engineer 1403 1404 # same as test_no_instance_key, but the PK columns 1405 # are absent. ensure an error is raised. 1406 sess = create_session() 1407 e1 = sess.query(Engineer).get(2) 1408 1409 sess.expire(e1, attribute_names=["name", "person_id"]) 1410 sess.expunge(e1) 1411 attributes.instance_state(e1).key = None 1412 assert "name" not in e1.__dict__ 1413 sess.add(e1) 1414 assert_raises(sa_exc.InvalidRequestError, getattr, e1, "name") 1415 1416 1417class ExpiredPendingTest(_fixtures.FixtureTest): 1418 run_define_tables = "once" 1419 run_setup_classes = "once" 1420 run_setup_mappers = None 1421 run_inserts = None 1422 1423 def test_expired_pending(self): 1424 users, Address, addresses, User = ( 1425 self.tables.users, 1426 self.classes.Address, 1427 self.tables.addresses, 1428 self.classes.User, 1429 ) 1430 1431 mapper( 1432 User, 1433 users, 1434 properties={"addresses": relationship(Address, backref="user")}, 1435 ) 1436 mapper(Address, addresses) 1437 1438 sess = create_session() 1439 a1 = Address(email_address="a1") 1440 sess.add(a1) 1441 sess.flush() 1442 1443 u1 = User(name="u1") 1444 a1.user = u1 1445 sess.flush() 1446 1447 # expire 'addresses'. backrefs 1448 # which attach to u1 will expect to be "pending" 1449 sess.expire(u1, ["addresses"]) 1450 1451 # attach an Address. now its "pending" 1452 # in user.addresses 1453 a2 = Address(email_address="a2") 1454 a2.user = u1 1455 1456 # expire u1.addresses again. this expires 1457 # "pending" as well. 1458 sess.expire(u1, ["addresses"]) 1459 1460 # insert a new row 1461 sess.execute( 1462 addresses.insert(), dict(email_address="a3", user_id=u1.id) 1463 ) 1464 1465 # only two addresses pulled from the DB, no "pending" 1466 assert len(u1.addresses) == 2 1467 1468 sess.flush() 1469 sess.expire_all() 1470 assert len(u1.addresses) == 3 1471 1472 1473class LifecycleTest(fixtures.MappedTest): 1474 @classmethod 1475 def define_tables(cls, metadata): 1476 Table( 1477 "data", 1478 metadata, 1479 Column( 1480 "id", Integer, primary_key=True, test_needs_autoincrement=True 1481 ), 1482 Column("data", String(30)), 1483 ) 1484 Table( 1485 "data_fetched", 1486 metadata, 1487 Column( 1488 "id", Integer, primary_key=True, test_needs_autoincrement=True 1489 ), 1490 Column("data", String(30), FetchedValue()), 1491 ) 1492 Table( 1493 "data_defer", 1494 metadata, 1495 Column( 1496 "id", Integer, primary_key=True, test_needs_autoincrement=True 1497 ), 1498 Column("data", String(30)), 1499 Column("data2", String(30)), 1500 ) 1501 1502 @classmethod 1503 def setup_classes(cls): 1504 class Data(cls.Comparable): 1505 pass 1506 1507 class DataFetched(cls.Comparable): 1508 pass 1509 1510 class DataDefer(cls.Comparable): 1511 pass 1512 1513 @classmethod 1514 def setup_mappers(cls): 1515 mapper(cls.classes.Data, cls.tables.data) 1516 mapper(cls.classes.DataFetched, cls.tables.data_fetched) 1517 mapper( 1518 cls.classes.DataDefer, 1519 cls.tables.data_defer, 1520 properties={"data": deferred(cls.tables.data_defer.c.data)}, 1521 ) 1522 1523 def test_attr_not_inserted(self): 1524 Data = self.classes.Data 1525 1526 sess = create_session() 1527 1528 d1 = Data() 1529 sess.add(d1) 1530 sess.flush() 1531 1532 # we didn't insert a value for 'data', 1533 # so its not in dict, but also when we hit it, it isn't 1534 # expired because there's no column default on it or anything like that 1535 assert "data" not in d1.__dict__ 1536 1537 def go(): 1538 eq_(d1.data, None) 1539 1540 self.assert_sql_count(testing.db, go, 0) 1541 1542 def test_attr_not_inserted_expired(self): 1543 Data = self.classes.Data 1544 1545 sess = create_session() 1546 1547 d1 = Data() 1548 sess.add(d1) 1549 sess.flush() 1550 1551 assert "data" not in d1.__dict__ 1552 1553 # with an expire, we emit 1554 sess.expire(d1) 1555 1556 def go(): 1557 eq_(d1.data, None) 1558 1559 self.assert_sql_count(testing.db, go, 1) 1560 1561 def test_attr_not_inserted_fetched(self): 1562 Data = self.classes.DataFetched 1563 1564 sess = create_session() 1565 1566 d1 = Data() 1567 sess.add(d1) 1568 sess.flush() 1569 1570 assert "data" not in d1.__dict__ 1571 1572 def go(): 1573 eq_(d1.data, None) 1574 1575 # this one is marked as "fetch" so we emit SQL 1576 self.assert_sql_count(testing.db, go, 1) 1577 1578 def test_cols_missing_in_load(self): 1579 Data = self.classes.Data 1580 1581 sess = create_session() 1582 1583 d1 = Data(data="d1") 1584 sess.add(d1) 1585 sess.flush() 1586 sess.close() 1587 1588 sess = create_session() 1589 d1 = sess.query(Data).from_statement(select([Data.id])).first() 1590 1591 # cols not present in the row are implicitly expired 1592 def go(): 1593 eq_(d1.data, "d1") 1594 1595 self.assert_sql_count(testing.db, go, 1) 1596 1597 def test_deferred_cols_missing_in_load_state_reset(self): 1598 Data = self.classes.DataDefer 1599 1600 sess = create_session() 1601 1602 d1 = Data(data="d1") 1603 sess.add(d1) 1604 sess.flush() 1605 sess.close() 1606 1607 sess = create_session() 1608 d1 = ( 1609 sess.query(Data) 1610 .from_statement(select([Data.id])) 1611 .options(undefer(Data.data)) 1612 .first() 1613 ) 1614 d1.data = "d2" 1615 1616 # the deferred loader has to clear out any state 1617 # on the col, including that 'd2' here 1618 d1 = sess.query(Data).populate_existing().first() 1619 1620 def go(): 1621 eq_(d1.data, "d1") 1622 1623 self.assert_sql_count(testing.db, go, 1) 1624 1625 1626class RefreshTest(_fixtures.FixtureTest): 1627 def test_refresh(self): 1628 users, Address, addresses, User = ( 1629 self.tables.users, 1630 self.classes.Address, 1631 self.tables.addresses, 1632 self.classes.User, 1633 ) 1634 1635 mapper( 1636 User, 1637 users, 1638 properties={ 1639 "addresses": relationship( 1640 mapper(Address, addresses), backref="user" 1641 ) 1642 }, 1643 ) 1644 s = create_session() 1645 u = s.query(User).get(7) 1646 u.name = "foo" 1647 a = Address() 1648 assert sa.orm.object_session(a) is None 1649 u.addresses.append(a) 1650 assert a.email_address is None 1651 assert id(a) in [id(x) for x in u.addresses] 1652 1653 s.refresh(u) 1654 1655 # its refreshed, so not dirty 1656 assert u not in s.dirty 1657 1658 # username is back to the DB 1659 assert u.name == "jack" 1660 1661 assert id(a) not in [id(x) for x in u.addresses] 1662 1663 u.name = "foo" 1664 u.addresses.append(a) 1665 # now its dirty 1666 assert u in s.dirty 1667 assert u.name == "foo" 1668 assert id(a) in [id(x) for x in u.addresses] 1669 s.expire(u) 1670 1671 # get the attribute, it refreshes 1672 assert u.name == "jack" 1673 assert id(a) not in [id(x) for x in u.addresses] 1674 1675 def test_persistence_check(self): 1676 users, User = self.tables.users, self.classes.User 1677 1678 mapper(User, users) 1679 s = create_session() 1680 u = s.query(User).get(7) 1681 s.expunge_all() 1682 assert_raises_message( 1683 sa_exc.InvalidRequestError, 1684 r"is not persistent within this Session", 1685 lambda: s.refresh(u), 1686 ) 1687 1688 def test_refresh_expired(self): 1689 User, users = self.classes.User, self.tables.users 1690 1691 mapper(User, users) 1692 s = create_session() 1693 u = s.query(User).get(7) 1694 s.expire(u) 1695 assert "name" not in u.__dict__ 1696 s.refresh(u) 1697 assert u.name == "jack" 1698 1699 def test_refresh_with_lazy(self): 1700 """test that when a lazy loader is set as a trigger on an object's 1701 attribute (at the attribute level, not the class level), a refresh() 1702 operation doesn't fire the lazy loader or create any problems""" 1703 1704 User, Address, addresses, users = ( 1705 self.classes.User, 1706 self.classes.Address, 1707 self.tables.addresses, 1708 self.tables.users, 1709 ) 1710 1711 s = create_session() 1712 mapper( 1713 User, 1714 users, 1715 properties={"addresses": relationship(mapper(Address, addresses))}, 1716 ) 1717 q = s.query(User).options(sa.orm.lazyload("addresses")) 1718 u = q.filter(users.c.id == 8).first() 1719 1720 def go(): 1721 s.refresh(u) 1722 1723 self.assert_sql_count(testing.db, go, 1) 1724 1725 def test_refresh_with_eager(self): 1726 """test that a refresh/expire operation loads rows properly and sends 1727 correct "isnew" state to eager loaders""" 1728 1729 users, Address, addresses, User = ( 1730 self.tables.users, 1731 self.classes.Address, 1732 self.tables.addresses, 1733 self.classes.User, 1734 ) 1735 1736 mapper( 1737 User, 1738 users, 1739 properties={ 1740 "addresses": relationship( 1741 mapper(Address, addresses), lazy="joined" 1742 ) 1743 }, 1744 ) 1745 1746 s = create_session() 1747 u = s.query(User).get(8) 1748 assert len(u.addresses) == 3 1749 s.refresh(u) 1750 assert len(u.addresses) == 3 1751 1752 s = create_session() 1753 u = s.query(User).get(8) 1754 assert len(u.addresses) == 3 1755 s.expire(u) 1756 assert len(u.addresses) == 3 1757 1758 def test_refresh_maintains_deferred_options(self): 1759 # testing a behavior that may have changed with 1760 # [ticket:3822] 1761 User, Address, Dingaling = self.classes("User", "Address", "Dingaling") 1762 users, addresses, dingalings = self.tables( 1763 "users", "addresses", "dingalings" 1764 ) 1765 1766 mapper(User, users, properties={"addresses": relationship(Address)}) 1767 1768 mapper( 1769 Address, 1770 addresses, 1771 properties={"dingalings": relationship(Dingaling)}, 1772 ) 1773 1774 mapper(Dingaling, dingalings) 1775 1776 s = create_session() 1777 q = ( 1778 s.query(User) 1779 .filter_by(name="fred") 1780 .options(sa.orm.lazyload("addresses").joinedload("dingalings")) 1781 ) 1782 1783 u1 = q.one() 1784 1785 # "addresses" is not present on u1, but when u1.addresses 1786 # lazy loads, it should also joinedload dingalings. This is 1787 # present in state.load_options and state.load_path. The 1788 # refresh operation should not reset these attributes. 1789 s.refresh(u1) 1790 1791 def go(): 1792 eq_( 1793 u1.addresses, 1794 [ 1795 Address( 1796 email_address="fred@fred.com", 1797 dingalings=[Dingaling(data="ding 2/5")], 1798 ) 1799 ], 1800 ) 1801 1802 self.assert_sql_count(testing.db, go, 1) 1803 1804 def test_refresh2(self): 1805 """test a hang condition that was occurring on expire/refresh""" 1806 1807 Address, addresses, users, User = ( 1808 self.classes.Address, 1809 self.tables.addresses, 1810 self.tables.users, 1811 self.classes.User, 1812 ) 1813 1814 s = create_session() 1815 mapper(Address, addresses) 1816 1817 mapper( 1818 User, 1819 users, 1820 properties=dict( 1821 addresses=relationship( 1822 Address, cascade="all, delete-orphan", lazy="joined" 1823 ) 1824 ), 1825 ) 1826 1827 u = User() 1828 u.name = "Justin" 1829 a = Address(id=10, email_address="lala") 1830 u.addresses.append(a) 1831 1832 s.add(u) 1833 s.flush() 1834 s.expunge_all() 1835 u = s.query(User).filter(User.name == "Justin").one() 1836 1837 s.expire(u) 1838 assert u.name == "Justin" 1839 1840 s.refresh(u) 1841