| 434 | cls.mapper_registry.map_imperatively(Address, addresses) |
| 435 | |
| 436 | def test_update(self): |
| 437 | User, Address = self.classes("User", "Address") |
| 438 | |
| 439 | s = Session(testing.db, future=True) |
| 440 | |
| 441 | def go(ids, values): |
| 442 | stmt = lambda_stmt(lambda: update(User).where(User.id.in_(ids))) |
| 443 | s.execute( |
| 444 | stmt, |
| 445 | values, |
| 446 | # note this currently just unrolls the lambda on the statement. |
| 447 | # so lambda caching for updates is not actually that useful |
| 448 | # unless synchronize_session is turned off. |
| 449 | # evaluate is similar just doesn't work for IN yet. |
| 450 | execution_options={"synchronize_session": "fetch"}, |
| 451 | ) |
| 452 | |
| 453 | go([1, 2], {"name": "jack2"}) |
| 454 | eq_( |
| 455 | s.execute(select(User.id, User.name).order_by(User.id)).all(), |
| 456 | [(1, "jack2"), (2, "jack2"), (3, "jill"), (4, "jane")], |
| 457 | ) |
| 458 | |
| 459 | go([3], {"name": "jane2"}) |
| 460 | eq_( |
| 461 | s.execute(select(User.id, User.name).order_by(User.id)).all(), |
| 462 | [(1, "jack2"), (2, "jack2"), (3, "jane2"), (4, "jane")], |
| 463 | ) |