1import sqlalchemy as sa 2from sqlalchemy import CheckConstraint 3from sqlalchemy import event 4from sqlalchemy import exc 5from sqlalchemy import ForeignKey 6from sqlalchemy import ForeignKeyConstraint 7from sqlalchemy import Index 8from sqlalchemy import inspect 9from sqlalchemy import Integer 10from sqlalchemy import String 11from sqlalchemy import testing 12from sqlalchemy import UniqueConstraint 13from sqlalchemy import util 14from sqlalchemy.ext.hybrid import hybrid_property 15from sqlalchemy.orm import backref 16from sqlalchemy.orm import class_mapper 17from sqlalchemy.orm import clear_mappers 18from sqlalchemy.orm import close_all_sessions 19from sqlalchemy.orm import column_property 20from sqlalchemy.orm import composite 21from sqlalchemy.orm import configure_mappers 22from sqlalchemy.orm import decl_base 23from sqlalchemy.orm import declarative_base 24from sqlalchemy.orm import declared_attr 25from sqlalchemy.orm import deferred 26from sqlalchemy.orm import descriptor_props 27from sqlalchemy.orm import exc as orm_exc 28from sqlalchemy.orm import joinedload 29from sqlalchemy.orm import mapper 30from sqlalchemy.orm import registry 31from sqlalchemy.orm import relationship 32from sqlalchemy.orm import Session 33from sqlalchemy.orm import synonym_for 34from sqlalchemy.orm.decl_api import DeclarativeMeta 35from sqlalchemy.orm.decl_base import _DeferredMapperConfig 36from sqlalchemy.orm.events import InstrumentationEvents 37from sqlalchemy.orm.events import MapperEvents 38from sqlalchemy.testing import assert_raises 39from sqlalchemy.testing import assert_raises_message 40from sqlalchemy.testing import assertions 41from sqlalchemy.testing import eq_ 42from sqlalchemy.testing import expect_warnings 43from sqlalchemy.testing import fixtures 44from sqlalchemy.testing import is_ 45from sqlalchemy.testing import mock 46from sqlalchemy.testing.fixtures import fixture_session 47from sqlalchemy.testing.schema import Column 48from sqlalchemy.testing.schema import Table 49from sqlalchemy.util import with_metaclass 50 51 52Base = None 53 54User = Address = None 55 56 57class DeclarativeTestBase( 58 fixtures.TestBase, 59 testing.AssertsExecutionResults, 60 testing.AssertsCompiledSQL, 61): 62 __dialect__ = "default" 63 64 base_style = "dynamic" 65 66 def setup_test(self): 67 global Base 68 69 if self.base_style == "dynamic": 70 Base = declarative_base() 71 elif self.base_style == "explicit": 72 mapper_registry = registry() 73 74 class Base(with_metaclass(DeclarativeMeta)): 75 __abstract__ = True 76 registry = mapper_registry 77 metadata = mapper_registry.metadata 78 79 def teardown_test(self): 80 close_all_sessions() 81 clear_mappers() 82 Base.metadata.drop_all(testing.db) 83 84 85@testing.combinations( 86 ("dynamic",), ("explicit",), argnames="base_style", id_="s" 87) 88class DeclarativeTest(DeclarativeTestBase): 89 def test_unbound_declarative_base(self): 90 Base = declarative_base() 91 92 class User(Base): 93 __tablename__ = "user" 94 id = Column(Integer, primary_key=True) 95 96 s = Session() 97 98 with testing.expect_raises(exc.UnboundExecutionError): 99 s.get_bind(User) 100 101 def test_unbound_cls_registry(self): 102 reg = registry() 103 104 Base = reg.generate_base() 105 106 class User(Base): 107 __tablename__ = "user" 108 id = Column(Integer, primary_key=True) 109 110 s = Session() 111 112 with testing.expect_raises(exc.UnboundExecutionError): 113 s.get_bind(User) 114 115 def test_basic(self): 116 class User(Base, fixtures.ComparableEntity): 117 __tablename__ = "users" 118 119 id = Column( 120 "id", Integer, primary_key=True, test_needs_autoincrement=True 121 ) 122 name = Column("name", String(50)) 123 addresses = relationship("Address", backref="user") 124 125 class Address(Base, fixtures.ComparableEntity): 126 __tablename__ = "addresses" 127 128 id = Column( 129 Integer, primary_key=True, test_needs_autoincrement=True 130 ) 131 email = Column(String(50), key="_email") 132 user_id = Column( 133 "user_id", Integer, ForeignKey("users.id"), key="_user_id" 134 ) 135 136 Base.metadata.create_all(testing.db) 137 138 eq_(Address.__table__.c["id"].name, "id") 139 eq_(Address.__table__.c["_email"].name, "email") 140 eq_(Address.__table__.c["_user_id"].name, "user_id") 141 142 u1 = User( 143 name="u1", addresses=[Address(email="one"), Address(email="two")] 144 ) 145 sess = fixture_session() 146 sess.add(u1) 147 sess.flush() 148 sess.expunge_all() 149 150 eq_( 151 sess.query(User).all(), 152 [ 153 User( 154 name="u1", 155 addresses=[Address(email="one"), Address(email="two")], 156 ) 157 ], 158 ) 159 160 a1 = sess.query(Address).filter(Address.email == "two").one() 161 eq_(a1, Address(email="two")) 162 eq_(a1.user, User(name="u1")) 163 164 def test_back_populates_setup(self): 165 class User(Base): 166 __tablename__ = "users" 167 168 id = Column("id", Integer, primary_key=True) 169 addresses = relationship("Address", back_populates="user") 170 171 class Address(Base): 172 __tablename__ = "addresses" 173 174 id = Column(Integer, primary_key=True) 175 user_id = Column( 176 "user_id", Integer, ForeignKey("users.id"), key="_user_id" 177 ) 178 user = relationship("User", back_populates="addresses") 179 180 configure_mappers() 181 182 assert ( 183 Address.__mapper__.attrs.user 184 in User.__mapper__.attrs.addresses._reverse_property 185 ) 186 assert ( 187 User.__mapper__.attrs.addresses 188 in Address.__mapper__.attrs.user._reverse_property 189 ) 190 191 def test_dispose_attrs(self): 192 reg = registry() 193 194 class Foo(object): 195 __tablename__ = "some_table" 196 197 id = Column(Integer, primary_key=True) 198 199 reg.mapped(Foo) 200 201 is_(Foo.__mapper__, class_mapper(Foo)) 202 is_(Foo.__table__, class_mapper(Foo).local_table) 203 204 clear_mappers() 205 206 assert not hasattr(Foo, "__mapper__") 207 assert not hasattr(Foo, "__table__") 208 209 from sqlalchemy.orm import clsregistry 210 211 assert clsregistry._key_is_empty( 212 "Foo", reg._class_registry, lambda cls: cls is Foo 213 ) 214 215 def test_deferred_reflection_default_error(self): 216 class MyExt(object): 217 @classmethod 218 def prepare(cls): 219 "sample prepare method" 220 to_map = _DeferredMapperConfig.classes_for_base(cls) 221 for thingy in to_map: 222 thingy.map({}) 223 224 @classmethod 225 def _sa_decl_prepare(cls): 226 pass 227 228 class User(MyExt, Base): 229 __tablename__ = "user" 230 id = Column(Integer, primary_key=True) 231 232 assert_raises_message( 233 orm_exc.UnmappedClassError, 234 "Class .*User has a deferred " 235 "mapping on it. It is not yet usable as a mapped class.", 236 fixture_session().query, 237 User, 238 ) 239 240 User.prepare() 241 242 self.assert_compile( 243 fixture_session().query(User), 244 'SELECT "user".id AS user_id FROM "user"', 245 ) 246 247 def test_unicode_string_resolve(self): 248 class User(Base, fixtures.ComparableEntity): 249 __tablename__ = "users" 250 251 id = Column( 252 "id", Integer, primary_key=True, test_needs_autoincrement=True 253 ) 254 name = Column("name", String(50)) 255 addresses = relationship(util.u("Address"), backref="user") 256 257 class Address(Base, fixtures.ComparableEntity): 258 __tablename__ = "addresses" 259 260 id = Column( 261 Integer, primary_key=True, test_needs_autoincrement=True 262 ) 263 email = Column(String(50), key="_email") 264 user_id = Column( 265 "user_id", Integer, ForeignKey("users.id"), key="_user_id" 266 ) 267 268 assert User.addresses.property.mapper.class_ is Address 269 270 def test_unicode_string_resolve_backref(self): 271 class User(Base, fixtures.ComparableEntity): 272 __tablename__ = "users" 273 274 id = Column( 275 "id", Integer, primary_key=True, test_needs_autoincrement=True 276 ) 277 name = Column("name", String(50)) 278 279 class Address(Base, fixtures.ComparableEntity): 280 __tablename__ = "addresses" 281 282 id = Column( 283 Integer, primary_key=True, test_needs_autoincrement=True 284 ) 285 email = Column(String(50), key="_email") 286 user_id = Column( 287 "user_id", Integer, ForeignKey("users.id"), key="_user_id" 288 ) 289 user = relationship( 290 User, 291 backref=backref("addresses", order_by=util.u("Address.email")), 292 ) 293 294 assert Address.user.property.mapper.class_ is User 295 296 def test_no_table(self): 297 def go(): 298 class User(Base): 299 id = Column("id", Integer, primary_key=True) 300 301 assert_raises_message( 302 sa.exc.InvalidRequestError, "does not have a __table__", go 303 ) 304 305 def test_table_args_empty_dict(self): 306 class MyModel(Base): 307 __tablename__ = "test" 308 id = Column(Integer, primary_key=True) 309 __table_args__ = {} 310 311 def test_table_args_empty_tuple(self): 312 class MyModel(Base): 313 __tablename__ = "test" 314 id = Column(Integer, primary_key=True) 315 __table_args__ = () 316 317 def test_cant_add_columns(self): 318 t = Table( 319 "t", 320 Base.metadata, 321 Column("id", Integer, primary_key=True), 322 Column("data", String), 323 ) 324 325 def go(): 326 class User(Base): 327 __table__ = t 328 foo = Column(Integer, primary_key=True) 329 330 # can't specify new columns not already in the table 331 332 assert_raises_message( 333 sa.exc.ArgumentError, 334 "Can't add additional column 'foo' when " "specifying __table__", 335 go, 336 ) 337 338 # regular re-mapping works tho 339 340 class Bar(Base): 341 __table__ = t 342 some_data = t.c.data 343 344 assert ( 345 class_mapper(Bar).get_property("some_data").columns[0] is t.c.data 346 ) 347 348 def test_lower_case_c_column_warning(self): 349 with assertions.expect_warnings( 350 r"Attribute 'x' on class <class .*Foo.* appears to be a " 351 r"non-schema 'sqlalchemy.sql.column\(\)' object; " 352 ): 353 354 class Foo(Base): 355 __tablename__ = "foo" 356 357 id = Column(Integer, primary_key=True) 358 x = sa.sql.expression.column(Integer) 359 y = Column(Integer) 360 361 class MyMixin(object): 362 x = sa.sql.expression.column(Integer) 363 y = Column(Integer) 364 365 with assertions.expect_warnings( 366 r"Attribute 'x' on class <class .*MyMixin.* appears to be a " 367 r"non-schema 'sqlalchemy.sql.column\(\)' object; " 368 ): 369 370 class Foo2(MyMixin, Base): 371 __tablename__ = "foo2" 372 373 id = Column(Integer, primary_key=True) 374 375 with assertions.expect_warnings( 376 r"Attribute 'x' on class <class .*Foo3.* appears to be a " 377 r"non-schema 'sqlalchemy.sql.column\(\)' object; " 378 ): 379 380 class Foo3(Base): 381 __tablename__ = "foo3" 382 383 id = Column(Integer, primary_key=True) 384 385 @declared_attr 386 def x(cls): 387 return sa.sql.expression.column(Integer) 388 389 y = Column(Integer) 390 391 with assertions.expect_warnings( 392 r"Attribute 'x' on class <class .*Foo4.* appears to be a " 393 r"non-schema 'sqlalchemy.sql.column\(\)' object; " 394 ): 395 396 class MyMixin2(object): 397 @declared_attr 398 def x(cls): 399 return sa.sql.expression.column(Integer) 400 401 y = Column(Integer) 402 403 class Foo4(MyMixin2, Base): 404 __tablename__ = "foo4" 405 406 id = Column(Integer, primary_key=True) 407 408 def test_column_named_twice(self): 409 def go(): 410 class Foo(Base): 411 __tablename__ = "foo" 412 413 id = Column(Integer, primary_key=True) 414 x = Column("x", Integer) 415 y = Column("x", Integer) 416 417 assert_raises_message( 418 sa.exc.SAWarning, 419 "On class 'Foo', Column object 'x' named directly multiple times, " 420 "only one will be used: x, y", 421 go, 422 ) 423 424 def test_column_repeated_under_prop(self): 425 def go(): 426 class Foo(Base): 427 __tablename__ = "foo" 428 429 id = Column(Integer, primary_key=True) 430 x = Column("x", Integer) 431 y = column_property(x) 432 z = Column("x", Integer) 433 434 assert_raises_message( 435 sa.exc.SAWarning, 436 "On class 'Foo', Column object 'x' named directly multiple times, " 437 "only one will be used: x, y, z", 438 go, 439 ) 440 441 def test_using_explicit_prop_in_schema_objects(self): 442 class Foo(Base): 443 __tablename__ = "foo" 444 445 id = Column(Integer, primary_key=True) 446 cprop = column_property(Column(Integer)) 447 448 __table_args__ = (UniqueConstraint(cprop),) 449 450 uq = [ 451 c 452 for c in Foo.__table__.constraints 453 if isinstance(c, UniqueConstraint) 454 ][0] 455 is_(uq.columns.cprop, Foo.__table__.c.cprop) 456 457 class Bar(Base): 458 __tablename__ = "bar" 459 460 id = Column(Integer, primary_key=True) 461 cprop = deferred(Column(Integer)) 462 463 __table_args__ = (CheckConstraint(cprop > sa.func.foo()),) 464 465 ck = [ 466 c 467 for c in Bar.__table__.constraints 468 if isinstance(c, CheckConstraint) 469 ][0] 470 is_(ck.columns.cprop, Bar.__table__.c.cprop) 471 472 if testing.requires.python3.enabled: 473 # test the existing failure case in case something changes 474 def go(): 475 class Bat(Base): 476 __tablename__ = "bat" 477 478 id = Column(Integer, primary_key=True) 479 cprop = deferred(Column(Integer)) 480 481 # we still can't do an expression like 482 # "cprop > 5" because the column property isn't 483 # a full blown column 484 485 __table_args__ = (CheckConstraint(cprop > 5),) 486 487 assert_raises(TypeError, go) 488 489 def test_relationship_level_msg_for_invalid_callable(self): 490 class A(Base): 491 __tablename__ = "a" 492 id = Column(Integer, primary_key=True) 493 494 class B(Base): 495 __tablename__ = "b" 496 id = Column(Integer, primary_key=True) 497 a_id = Column(Integer, ForeignKey("a.id")) 498 a = relationship("a") 499 500 assert_raises_message( 501 sa.exc.ArgumentError, 502 "relationship 'a' expects a class or a mapper " 503 "argument .received: .*Table", 504 configure_mappers, 505 ) 506 507 def test_relationship_level_msg_for_invalid_object(self): 508 class A(Base): 509 __tablename__ = "a" 510 id = Column(Integer, primary_key=True) 511 512 class B(Base): 513 __tablename__ = "b" 514 id = Column(Integer, primary_key=True) 515 a_id = Column(Integer, ForeignKey("a.id")) 516 a = relationship(A.__table__) 517 518 assert_raises_message( 519 sa.exc.ArgumentError, 520 "relationship 'a' expects a class or a mapper " 521 "argument .received: .*Table", 522 configure_mappers, 523 ) 524 525 def test_difficult_class(self): 526 """test no getattr() errors with a customized class""" 527 528 # metaclass to mock the way zope.interface breaks getattr() 529 class BrokenMeta(type): 530 def __getattribute__(self, attr): 531 if attr == "xyzzy": 532 raise AttributeError("xyzzy") 533 else: 534 return object.__getattribute__(self, attr) 535 536 # even though this class has an xyzzy attribute, getattr(cls,"xyzzy") 537 # fails 538 class BrokenParent(with_metaclass(BrokenMeta)): 539 xyzzy = "magic" 540 541 # _as_declarative() inspects obj.__class__.__bases__ 542 class User(BrokenParent, fixtures.ComparableEntity): 543 __tablename__ = "users" 544 id = Column( 545 "id", Integer, primary_key=True, test_needs_autoincrement=True 546 ) 547 name = Column("name", String(50)) 548 549 reg = registry(metadata=Base.metadata) 550 551 reg.map_declaratively(User) 552 553 def test_reserved_identifiers(self): 554 def go1(): 555 class User1(Base): 556 __tablename__ = "user1" 557 id = Column(Integer, primary_key=True) 558 metadata = Column(Integer) 559 560 def go2(): 561 class User2(Base): 562 __tablename__ = "user2" 563 id = Column(Integer, primary_key=True) 564 metadata = relationship("Address") 565 566 for go in (go1, go2): 567 assert_raises_message( 568 exc.InvalidRequestError, 569 "Attribute name 'metadata' is reserved " 570 "for the MetaData instance when using a " 571 "declarative base class.", 572 go, 573 ) 574 575 def test_undefer_column_name(self): 576 # TODO: not sure if there was an explicit 577 # test for this elsewhere 578 foo = Column(Integer) 579 eq_(str(foo), "(no name)") 580 eq_(foo.key, None) 581 eq_(foo.name, None) 582 decl_base._undefer_column_name("foo", foo) 583 eq_(str(foo), "foo") 584 eq_(foo.key, "foo") 585 eq_(foo.name, "foo") 586 587 def test_recompile_on_othermapper(self): 588 """declarative version of the same test in mappers.py""" 589 590 class User(Base): 591 __tablename__ = "users" 592 593 id = Column("id", Integer, primary_key=True) 594 name = Column("name", String(50)) 595 596 class Address(Base): 597 __tablename__ = "addresses" 598 599 id = Column("id", Integer, primary_key=True) 600 email = Column("email", String(50)) 601 user_id = Column("user_id", Integer, ForeignKey("users.id")) 602 user = relationship( 603 "User", primaryjoin=user_id == User.id, backref="addresses" 604 ) 605 606 assert User.__mapper__.registry._new_mappers is True 607 u = User() # noqa 608 assert User.addresses 609 assert User.__mapper__.registry._new_mappers is False 610 611 def test_string_dependency_resolution(self): 612 class User(Base, fixtures.ComparableEntity): 613 614 __tablename__ = "users" 615 id = Column( 616 Integer, primary_key=True, test_needs_autoincrement=True 617 ) 618 name = Column(String(50)) 619 addresses = relationship( 620 "Address", 621 order_by="desc(Address.email)", 622 primaryjoin="User.id==Address.user_id", 623 foreign_keys="[Address.user_id]", 624 backref=backref( 625 "user", 626 primaryjoin="User.id==Address.user_id", 627 foreign_keys="[Address.user_id]", 628 ), 629 ) 630 631 class Address(Base, fixtures.ComparableEntity): 632 633 __tablename__ = "addresses" 634 id = Column( 635 Integer, primary_key=True, test_needs_autoincrement=True 636 ) 637 email = Column(String(50)) 638 user_id = Column(Integer) # note no foreign key 639 640 Base.metadata.create_all(testing.db) 641 sess = fixture_session() 642 u1 = User( 643 name="ed", 644 addresses=[ 645 Address(email="abc"), 646 Address(email="def"), 647 Address(email="xyz"), 648 ], 649 ) 650 sess.add(u1) 651 sess.flush() 652 sess.expunge_all() 653 eq_( 654 sess.query(User).filter(User.name == "ed").one(), 655 User( 656 name="ed", 657 addresses=[ 658 Address(email="xyz"), 659 Address(email="def"), 660 Address(email="abc"), 661 ], 662 ), 663 ) 664 665 class Foo(Base, fixtures.ComparableEntity): 666 667 __tablename__ = "foo" 668 id = Column(Integer, primary_key=True) 669 rel = relationship("User", primaryjoin="User.addresses==Foo.id") 670 671 assert_raises_message( 672 exc.InvalidRequestError, 673 "'addresses' is not an instance of " "ColumnProperty", 674 configure_mappers, 675 ) 676 677 def test_string_dependency_resolution_synonym(self): 678 class User(Base, fixtures.ComparableEntity): 679 680 __tablename__ = "users" 681 id = Column( 682 Integer, primary_key=True, test_needs_autoincrement=True 683 ) 684 name = Column(String(50)) 685 686 Base.metadata.create_all(testing.db) 687 sess = fixture_session() 688 u1 = User(name="ed") 689 sess.add(u1) 690 sess.flush() 691 sess.expunge_all() 692 eq_(sess.query(User).filter(User.name == "ed").one(), User(name="ed")) 693 694 class Foo(Base, fixtures.ComparableEntity): 695 696 __tablename__ = "foo" 697 id = Column(Integer, primary_key=True) 698 _user_id = Column(Integer) 699 rel = relationship( 700 "User", 701 uselist=False, 702 foreign_keys=[User.id], 703 primaryjoin="Foo.user_id==User.id", 704 ) 705 706 @synonym_for("_user_id") 707 @property 708 def user_id(self): 709 return self._user_id 710 711 foo = Foo() 712 foo.rel = u1 713 assert foo.rel == u1 714 715 def test_string_dependency_resolution_orm_descriptor(self): 716 from sqlalchemy.ext.hybrid import hybrid_property 717 718 class User(Base): 719 __tablename__ = "user" 720 id = Column(Integer, primary_key=True) 721 firstname = Column(String(50)) 722 lastname = Column(String(50)) 723 game_id = Column(Integer, ForeignKey("game.id")) 724 725 @hybrid_property 726 def fullname(self): 727 return self.firstname + " " + self.lastname 728 729 class Game(Base): 730 __tablename__ = "game" 731 id = Column(Integer, primary_key=True) 732 name = Column(String(50)) 733 users = relationship("User", order_by="User.fullname") 734 735 s = fixture_session() 736 self.assert_compile( 737 s.query(Game).options(joinedload(Game.users)), 738 "SELECT game.id AS game_id, game.name AS game_name, " 739 "user_1.id AS user_1_id, user_1.firstname AS user_1_firstname, " 740 "user_1.lastname AS user_1_lastname, " 741 "user_1.game_id AS user_1_game_id " 742 'FROM game LEFT OUTER JOIN "user" AS user_1 ON game.id = ' 743 "user_1.game_id ORDER BY " 744 "user_1.firstname || :firstname_1 || user_1.lastname", 745 ) 746 747 def test_string_dependency_resolution_asselectable(self): 748 class A(Base): 749 __tablename__ = "a" 750 751 id = Column(Integer, primary_key=True) 752 b_id = Column(ForeignKey("b.id")) 753 754 d = relationship( 755 "D", 756 secondary="join(B, D, B.d_id == D.id)." 757 "join(C, C.d_id == D.id)", 758 primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)", 759 secondaryjoin="D.id == B.d_id", 760 ) 761 762 class B(Base): 763 __tablename__ = "b" 764 765 id = Column(Integer, primary_key=True) 766 d_id = Column(ForeignKey("d.id")) 767 768 class C(Base): 769 __tablename__ = "c" 770 771 id = Column(Integer, primary_key=True) 772 a_id = Column(ForeignKey("a.id")) 773 d_id = Column(ForeignKey("d.id")) 774 775 class D(Base): 776 __tablename__ = "d" 777 778 id = Column(Integer, primary_key=True) 779 780 s = fixture_session() 781 self.assert_compile( 782 s.query(A).join(A.d), 783 "SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN " 784 "(b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id " 785 "JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id " 786 "AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id", 787 ) 788 789 def test_string_dependency_resolution_no_table(self): 790 class User(Base, fixtures.ComparableEntity): 791 __tablename__ = "users" 792 id = Column( 793 Integer, primary_key=True, test_needs_autoincrement=True 794 ) 795 name = Column(String(50)) 796 797 class Bar(Base, fixtures.ComparableEntity): 798 __tablename__ = "bar" 799 id = Column(Integer, primary_key=True) 800 rel = relationship("User", primaryjoin="User.id==Bar.__table__.id") 801 802 assert_raises_message( 803 AttributeError, 804 "does not have a mapped column named " "'__table__'", 805 configure_mappers, 806 ) 807 808 def test_string_w_pj_annotations(self): 809 class User(Base, fixtures.ComparableEntity): 810 __tablename__ = "users" 811 id = Column( 812 Integer, primary_key=True, test_needs_autoincrement=True 813 ) 814 name = Column(String(50)) 815 816 class Address(Base, fixtures.ComparableEntity): 817 818 __tablename__ = "addresses" 819 id = Column( 820 Integer, primary_key=True, test_needs_autoincrement=True 821 ) 822 email = Column(String(50)) 823 user_id = Column(Integer) 824 user = relationship( 825 "User", primaryjoin="remote(User.id)==foreign(Address.user_id)" 826 ) 827 828 eq_( 829 Address.user.property._join_condition.local_remote_pairs, 830 [(Address.__table__.c.user_id, User.__table__.c.id)], 831 ) 832 833 def test_string_dependency_resolution_no_magic(self): 834 """test that full tinkery expressions work as written""" 835 836 class User(Base, fixtures.ComparableEntity): 837 838 __tablename__ = "users" 839 id = Column(Integer, primary_key=True) 840 addresses = relationship( 841 "Address", 842 primaryjoin="User.id==Address.user_id.prop.columns[0]", 843 ) 844 845 class Address(Base, fixtures.ComparableEntity): 846 847 __tablename__ = "addresses" 848 id = Column(Integer, primary_key=True) 849 user_id = Column(Integer, ForeignKey("users.id")) 850 851 configure_mappers() 852 eq_( 853 str(User.addresses.prop.primaryjoin), 854 "users.id = addresses.user_id", 855 ) 856 857 def test_string_dependency_resolution_module_qualified(self): 858 class User(Base, fixtures.ComparableEntity): 859 860 __tablename__ = "users" 861 id = Column(Integer, primary_key=True) 862 addresses = relationship( 863 "%s.Address" % __name__, 864 primaryjoin="%s.User.id==%s.Address.user_id.prop.columns[0]" 865 % (__name__, __name__), 866 ) 867 868 class Address(Base, fixtures.ComparableEntity): 869 870 __tablename__ = "addresses" 871 id = Column(Integer, primary_key=True) 872 user_id = Column(Integer, ForeignKey("users.id")) 873 874 configure_mappers() 875 eq_( 876 str(User.addresses.prop.primaryjoin), 877 "users.id = addresses.user_id", 878 ) 879 880 def test_string_dependency_resolution_in_backref(self): 881 class User(Base, fixtures.ComparableEntity): 882 883 __tablename__ = "users" 884 id = Column(Integer, primary_key=True) 885 name = Column(String(50)) 886 addresses = relationship( 887 "Address", 888 primaryjoin="User.id==Address.user_id", 889 backref="user", 890 ) 891 892 class Address(Base, fixtures.ComparableEntity): 893 894 __tablename__ = "addresses" 895 id = Column(Integer, primary_key=True) 896 email = Column(String(50)) 897 user_id = Column(Integer, ForeignKey("users.id")) 898 899 configure_mappers() 900 eq_( 901 str(User.addresses.property.primaryjoin), 902 str(Address.user.property.primaryjoin), 903 ) 904 905 def test_string_dependency_resolution_tables(self): 906 class User(Base, fixtures.ComparableEntity): 907 908 __tablename__ = "users" 909 id = Column(Integer, primary_key=True) 910 name = Column(String(50)) 911 props = relationship( 912 "Prop", 913 secondary="user_to_prop", 914 primaryjoin="User.id==user_to_prop.c.user_id", 915 secondaryjoin="user_to_prop.c.prop_id==Prop.id", 916 backref="users", 917 ) 918 919 class Prop(Base, fixtures.ComparableEntity): 920 921 __tablename__ = "props" 922 id = Column(Integer, primary_key=True) 923 name = Column(String(50)) 924 925 user_to_prop = Table( 926 "user_to_prop", 927 Base.metadata, 928 Column("user_id", Integer, ForeignKey("users.id")), 929 Column("prop_id", Integer, ForeignKey("props.id")), 930 ) 931 932 configure_mappers() 933 assert ( 934 class_mapper(User).get_property("props").secondary is user_to_prop 935 ) 936 937 def test_string_dependency_resolution_table_over_class(self): 938 # test for second half of #5774 939 class User(Base, fixtures.ComparableEntity): 940 941 __tablename__ = "users" 942 id = Column(Integer, primary_key=True) 943 name = Column(String(50)) 944 props = relationship( 945 "Prop", 946 secondary="Secondary", 947 backref="users", 948 ) 949 950 class Prop(Base, fixtures.ComparableEntity): 951 952 __tablename__ = "props" 953 id = Column(Integer, primary_key=True) 954 name = Column(String(50)) 955 956 # class name and table name match 957 class Secondary(Base): 958 __tablename__ = "Secondary" 959 user_id = Column(Integer, ForeignKey("users.id"), primary_key=True) 960 prop_id = Column(Integer, ForeignKey("props.id"), primary_key=True) 961 962 configure_mappers() 963 assert ( 964 class_mapper(User).get_property("props").secondary 965 is Secondary.__table__ 966 ) 967 968 def test_string_dependency_resolution_class_over_table(self): 969 # test for second half of #5774 970 class User(Base, fixtures.ComparableEntity): 971 972 __tablename__ = "users" 973 id = Column(Integer, primary_key=True) 974 name = Column(String(50)) 975 secondary = relationship( 976 "Secondary", 977 ) 978 979 # class name and table name match 980 class Secondary(Base): 981 __tablename__ = "Secondary" 982 user_id = Column(Integer, ForeignKey("users.id"), primary_key=True) 983 984 configure_mappers() 985 assert ( 986 class_mapper(User).get_property("secondary").mapper 987 is Secondary.__mapper__ 988 ) 989 990 def test_string_dependency_resolution_schemas(self): 991 Base = declarative_base() 992 993 class User(Base): 994 995 __tablename__ = "users" 996 __table_args__ = {"schema": "fooschema"} 997 998 id = Column(Integer, primary_key=True) 999 name = Column(String(50)) 1000 props = relationship( 1001 "Prop", 1002 secondary="fooschema.user_to_prop", 1003 primaryjoin="User.id==fooschema.user_to_prop.c.user_id", 1004 secondaryjoin="fooschema.user_to_prop.c.prop_id==Prop.id", 1005 backref="users", 1006 ) 1007 1008 class Prop(Base): 1009 1010 __tablename__ = "props" 1011 __table_args__ = {"schema": "fooschema"} 1012 1013 id = Column(Integer, primary_key=True) 1014 name = Column(String(50)) 1015 1016 user_to_prop = Table( 1017 "user_to_prop", 1018 Base.metadata, 1019 Column("user_id", Integer, ForeignKey("fooschema.users.id")), 1020 Column("prop_id", Integer, ForeignKey("fooschema.props.id")), 1021 schema="fooschema", 1022 ) 1023 configure_mappers() 1024 1025 assert ( 1026 class_mapper(User).get_property("props").secondary is user_to_prop 1027 ) 1028 1029 def test_string_dependency_resolution_annotations(self): 1030 Base = declarative_base() 1031 1032 class Parent(Base): 1033 __tablename__ = "parent" 1034 id = Column(Integer, primary_key=True) 1035 name = Column(String) 1036 children = relationship( 1037 "Child", 1038 primaryjoin="Parent.name==" 1039 "remote(foreign(func.lower(Child.name_upper)))", 1040 ) 1041 1042 class Child(Base): 1043 __tablename__ = "child" 1044 id = Column(Integer, primary_key=True) 1045 name_upper = Column(String) 1046 1047 configure_mappers() 1048 eq_( 1049 Parent.children.property._calculated_foreign_keys, 1050 set([Child.name_upper.property.columns[0]]), 1051 ) 1052 1053 def test_class_has_registry_attr(self): 1054 existing_registry = Base.registry 1055 1056 class A(Base): 1057 __tablename__ = "a" 1058 1059 registry = {"foo": "bar"} 1060 id = Column(Integer, primary_key=True) 1061 data = Column(String) 1062 1063 class SubA(A): 1064 pass 1065 1066 is_(Base.registry, existing_registry) 1067 is_(inspect(A).registry, existing_registry) 1068 eq_(A.registry, {"foo": "bar"}) 1069 1070 is_(inspect(SubA).registry, existing_registry) 1071 eq_(SubA.registry, {"foo": "bar"}) 1072 1073 def test_class_does_not_have_registry_attr(self): 1074 with assertions.expect_raises_message( 1075 exc.InvalidRequestError, 1076 r"Declarative base class has no 'registry' attribute, or " 1077 r"registry is not a sqlalchemy.orm.registry\(\) object", 1078 ): 1079 1080 class Base(with_metaclass(DeclarativeMeta)): 1081 metadata = sa.MetaData() 1082 1083 def test_shared_class_registry(self): 1084 reg = {} 1085 Base1 = declarative_base(class_registry=reg) 1086 Base2 = declarative_base(class_registry=reg) 1087 1088 class A(Base1): 1089 __tablename__ = "a" 1090 id = Column(Integer, primary_key=True) 1091 1092 class B(Base2): 1093 __tablename__ = "b" 1094 id = Column(Integer, primary_key=True) 1095 aid = Column(Integer, ForeignKey(A.id)) 1096 as_ = relationship("A") 1097 1098 assert B.as_.property.mapper.class_ is A 1099 1100 def test_uncompiled_attributes_in_relationship(self): 1101 class Address(Base, fixtures.ComparableEntity): 1102 1103 __tablename__ = "addresses" 1104 id = Column( 1105 Integer, primary_key=True, test_needs_autoincrement=True 1106 ) 1107 email = Column(String(50)) 1108 user_id = Column(Integer, ForeignKey("users.id")) 1109 1110 class User(Base, fixtures.ComparableEntity): 1111 1112 __tablename__ = "users" 1113 id = Column( 1114 Integer, primary_key=True, test_needs_autoincrement=True 1115 ) 1116 name = Column(String(50)) 1117 addresses = relationship( 1118 "Address", 1119 order_by=Address.email, 1120 foreign_keys=Address.user_id, 1121 remote_side=Address.user_id, 1122 ) 1123 1124 # get the mapper for User. User mapper will compile, 1125 # "addresses" relationship will call upon Address.user_id for 1126 # its clause element. Address.user_id is a _CompileOnAttr, 1127 # which then calls class_mapper(Address). But ! We're already 1128 # "in compilation", but class_mapper(Address) needs to 1129 # initialize regardless, or COA's assertion fails and things 1130 # generally go downhill from there. 1131 1132 class_mapper(User) 1133 Base.metadata.create_all(testing.db) 1134 sess = fixture_session() 1135 u1 = User( 1136 name="ed", 1137 addresses=[ 1138 Address(email="abc"), 1139 Address(email="xyz"), 1140 Address(email="def"), 1141 ], 1142 ) 1143 sess.add(u1) 1144 sess.flush() 1145 sess.expunge_all() 1146 eq_( 1147 sess.query(User).filter(User.name == "ed").one(), 1148 User( 1149 name="ed", 1150 addresses=[ 1151 Address(email="abc"), 1152 Address(email="def"), 1153 Address(email="xyz"), 1154 ], 1155 ), 1156 ) 1157 1158 def test_nice_dependency_error(self): 1159 class User(Base): 1160 1161 __tablename__ = "users" 1162 id = Column("id", Integer, primary_key=True) 1163 addresses = relationship("Address") 1164 1165 class Address(Base): 1166 1167 __tablename__ = "addresses" 1168 id = Column(Integer, primary_key=True) 1169 foo = sa.orm.column_property(User.id == 5) 1170 1171 # this used to raise an error when accessing User.id but that's 1172 # no longer the case since we got rid of _CompileOnAttr. 1173 1174 assert_raises(sa.exc.ArgumentError, configure_mappers) 1175 1176 def test_nice_dependency_error_works_with_hasattr(self): 1177 class User(Base): 1178 1179 __tablename__ = "users" 1180 id = Column("id", Integer, primary_key=True) 1181 addresses = relationship("Address") 1182 1183 # hasattr() on a compile-loaded attribute 1184 try: 1185 hasattr(User.addresses, "property") 1186 except exc.InvalidRequestError: 1187 assert sa.util.compat.py3k 1188 1189 # the exception is preserved. Remains the 1190 # same through repeated calls. 1191 for i in range(3): 1192 assert_raises_message( 1193 sa.exc.InvalidRequestError, 1194 "^One or more mappers failed to initialize" 1195 " - can't proceed with initialization of other mappers. " 1196 r"Triggering mapper: 'mapped class User->users'. " 1197 "Original exception was: When initializing.*", 1198 configure_mappers, 1199 ) 1200 1201 def test_custom_base(self): 1202 class MyBase(object): 1203 def foobar(self): 1204 return "foobar" 1205 1206 Base = declarative_base(cls=MyBase) 1207 assert hasattr(Base, "metadata") 1208 assert Base().foobar() == "foobar" 1209 1210 def test_uses_get_on_class_col_fk(self): 1211 1212 # test [ticket:1492] 1213 1214 class Topic(Base): 1215 1216 __tablename__ = "topic" 1217 id = Column( 1218 Integer, primary_key=True, test_needs_autoincrement=True 1219 ) 1220 1221 class Detail(Base): 1222 1223 __tablename__ = "detail" 1224 id = Column( 1225 Integer, primary_key=True, test_needs_autoincrement=True 1226 ) 1227 topic_id = Column(None, ForeignKey(Topic.id)) 1228 topic = relationship(Topic) 1229 1230 Base.metadata.create_all(testing.db) 1231 configure_mappers() 1232 assert class_mapper(Detail).get_property("topic").strategy.use_get 1233 t1 = Topic() 1234 d1 = Detail(topic=t1) 1235 sess = fixture_session() 1236 sess.add(d1) 1237 sess.flush() 1238 sess.expunge_all() 1239 d1 = sess.query(Detail).first() 1240 t1 = sess.query(Topic).first() 1241 1242 def go(): 1243 assert d1.topic 1244 1245 self.assert_sql_count(testing.db, go, 0) 1246 1247 def test_index_doesnt_compile(self): 1248 class User(Base): 1249 __tablename__ = "users" 1250 id = Column("id", Integer, primary_key=True) 1251 name = Column("name", String(50)) 1252 error = relationship("Address") 1253 1254 i = Index("my_index", User.name) 1255 1256 # compile fails due to the nonexistent Addresses relationship 1257 assert_raises(sa.exc.InvalidRequestError, configure_mappers) 1258 1259 # index configured 1260 assert i in User.__table__.indexes 1261 assert User.__table__.c.id not in set(i.columns) 1262 assert User.__table__.c.name in set(i.columns) 1263 1264 # tables create fine 1265 Base.metadata.create_all(testing.db) 1266 1267 def test_add_prop(self): 1268 class User(Base, fixtures.ComparableEntity): 1269 1270 __tablename__ = "users" 1271 id = Column( 1272 "id", Integer, primary_key=True, test_needs_autoincrement=True 1273 ) 1274 1275 User.name = Column("name", String(50)) 1276 User.addresses = relationship("Address", backref="user") 1277 1278 class Address(Base, fixtures.ComparableEntity): 1279 1280 __tablename__ = "addresses" 1281 id = Column( 1282 Integer, primary_key=True, test_needs_autoincrement=True 1283 ) 1284 1285 Address.email = Column(String(50), key="_email") 1286 Address.user_id = Column( 1287 "user_id", Integer, ForeignKey("users.id"), key="_user_id" 1288 ) 1289 Base.metadata.create_all(testing.db) 1290 eq_(Address.__table__.c["id"].name, "id") 1291 eq_(Address.__table__.c["_email"].name, "email") 1292 eq_(Address.__table__.c["_user_id"].name, "user_id") 1293 u1 = User( 1294 name="u1", addresses=[Address(email="one"), Address(email="two")] 1295 ) 1296 sess = fixture_session() 1297 sess.add(u1) 1298 sess.flush() 1299 sess.expunge_all() 1300 eq_( 1301 sess.query(User).all(), 1302 [ 1303 User( 1304 name="u1", 1305 addresses=[Address(email="one"), Address(email="two")], 1306 ) 1307 ], 1308 ) 1309 a1 = sess.query(Address).filter(Address.email == "two").one() 1310 eq_(a1, Address(email="two")) 1311 eq_(a1.user, User(name="u1")) 1312 1313 def test_alt_name_attr_subclass_column_inline(self): 1314 # [ticket:2900] 1315 class A(Base): 1316 __tablename__ = "a" 1317 id = Column("id", Integer, primary_key=True) 1318 data = Column("data") 1319 1320 class ASub(A): 1321 brap = A.data 1322 1323 assert ASub.brap.property is A.data.property 1324 assert isinstance( 1325 ASub.brap.original_property, descriptor_props.SynonymProperty 1326 ) 1327 1328 def test_alt_name_attr_subclass_relationship_inline(self): 1329 # [ticket:2900] 1330 class A(Base): 1331 __tablename__ = "a" 1332 id = Column("id", Integer, primary_key=True) 1333 b_id = Column(Integer, ForeignKey("b.id")) 1334 b = relationship("B", backref="as_") 1335 1336 class B(Base): 1337 __tablename__ = "b" 1338 id = Column("id", Integer, primary_key=True) 1339 1340 configure_mappers() 1341 1342 class ASub(A): 1343 brap = A.b 1344 1345 assert ASub.brap.property is A.b.property 1346 assert isinstance( 1347 ASub.brap.original_property, descriptor_props.SynonymProperty 1348 ) 1349 ASub(brap=B()) 1350 1351 def test_alt_name_attr_subclass_column_attrset(self): 1352 # [ticket:2900] 1353 class A(Base): 1354 __tablename__ = "a" 1355 id = Column("id", Integer, primary_key=True) 1356 data = Column("data") 1357 1358 A.brap = A.data 1359 assert A.brap.property is A.data.property 1360 assert isinstance( 1361 A.brap.original_property, descriptor_props.SynonymProperty 1362 ) 1363 1364 def test_alt_name_attr_subclass_relationship_attrset(self): 1365 # [ticket:2900] 1366 class A(Base): 1367 __tablename__ = "a" 1368 id = Column("id", Integer, primary_key=True) 1369 b_id = Column(Integer, ForeignKey("b.id")) 1370 b = relationship("B", backref="as_") 1371 1372 A.brap = A.b 1373 1374 class B(Base): 1375 __tablename__ = "b" 1376 id = Column("id", Integer, primary_key=True) 1377 1378 assert A.brap.property is A.b.property 1379 assert isinstance( 1380 A.brap.original_property, descriptor_props.SynonymProperty 1381 ) 1382 A(brap=B()) 1383 1384 def test_eager_order_by(self): 1385 class Address(Base, fixtures.ComparableEntity): 1386 1387 __tablename__ = "addresses" 1388 id = Column( 1389 "id", Integer, primary_key=True, test_needs_autoincrement=True 1390 ) 1391 email = Column("email", String(50)) 1392 user_id = Column("user_id", Integer, ForeignKey("users.id")) 1393 1394 class User(Base, fixtures.ComparableEntity): 1395 1396 __tablename__ = "users" 1397 id = Column( 1398 "id", Integer, primary_key=True, test_needs_autoincrement=True 1399 ) 1400 name = Column("name", String(50)) 1401 addresses = relationship("Address", order_by=Address.email) 1402 1403 Base.metadata.create_all(testing.db) 1404 u1 = User( 1405 name="u1", addresses=[Address(email="two"), Address(email="one")] 1406 ) 1407 sess = fixture_session() 1408 sess.add(u1) 1409 sess.flush() 1410 sess.expunge_all() 1411 eq_( 1412 sess.query(User).options(joinedload(User.addresses)).all(), 1413 [ 1414 User( 1415 name="u1", 1416 addresses=[Address(email="one"), Address(email="two")], 1417 ) 1418 ], 1419 ) 1420 1421 def test_order_by_multi(self): 1422 class Address(Base, fixtures.ComparableEntity): 1423 1424 __tablename__ = "addresses" 1425 id = Column( 1426 "id", Integer, primary_key=True, test_needs_autoincrement=True 1427 ) 1428 email = Column("email", String(50)) 1429 user_id = Column("user_id", Integer, ForeignKey("users.id")) 1430 1431 class User(Base, fixtures.ComparableEntity): 1432 1433 __tablename__ = "users" 1434 id = Column( 1435 "id", Integer, primary_key=True, test_needs_autoincrement=True 1436 ) 1437 name = Column("name", String(50)) 1438 addresses = relationship( 1439 "Address", order_by=(Address.email, Address.id) 1440 ) 1441 1442 Base.metadata.create_all(testing.db) 1443 u1 = User( 1444 name="u1", addresses=[Address(email="two"), Address(email="one")] 1445 ) 1446 sess = fixture_session() 1447 sess.add(u1) 1448 sess.flush() 1449 sess.expunge_all() 1450 u = sess.query(User).filter(User.name == "u1").one() 1451 u.addresses 1452 1453 def test_as_declarative(self): 1454 class User(fixtures.ComparableEntity): 1455 1456 __tablename__ = "users" 1457 id = Column( 1458 "id", Integer, primary_key=True, test_needs_autoincrement=True 1459 ) 1460 name = Column("name", String(50)) 1461 addresses = relationship("Address", backref="user") 1462 1463 class Address(fixtures.ComparableEntity): 1464 1465 __tablename__ = "addresses" 1466 id = Column( 1467 "id", Integer, primary_key=True, test_needs_autoincrement=True 1468 ) 1469 email = Column("email", String(50)) 1470 user_id = Column("user_id", Integer, ForeignKey("users.id")) 1471 1472 reg = registry(metadata=Base.metadata) 1473 reg.mapped(User) 1474 reg.mapped(Address) 1475 reg.metadata.create_all(testing.db) 1476 u1 = User( 1477 name="u1", addresses=[Address(email="one"), Address(email="two")] 1478 ) 1479 with Session(testing.db) as sess: 1480 sess.add(u1) 1481 sess.commit() 1482 with Session(testing.db) as sess: 1483 eq_( 1484 sess.query(User).all(), 1485 [ 1486 User( 1487 name="u1", 1488 addresses=[Address(email="one"), Address(email="two")], 1489 ) 1490 ], 1491 ) 1492 1493 def test_custom_mapper_attribute(self): 1494 def mymapper(cls, tbl, **kwargs): 1495 m = sa.orm.mapper(cls, tbl, **kwargs) 1496 m.CHECK = True 1497 return m 1498 1499 base = declarative_base() 1500 1501 class Foo(base): 1502 __tablename__ = "foo" 1503 __mapper_cls__ = mymapper 1504 id = Column(Integer, primary_key=True) 1505 1506 eq_(Foo.__mapper__.CHECK, True) 1507 1508 def test_custom_mapper_argument(self): 1509 def mymapper(cls, tbl, **kwargs): 1510 m = sa.orm.mapper(cls, tbl, **kwargs) 1511 m.CHECK = True 1512 return m 1513 1514 base = declarative_base(mapper=mymapper) 1515 1516 class Foo(base): 1517 __tablename__ = "foo" 1518 id = Column(Integer, primary_key=True) 1519 1520 eq_(Foo.__mapper__.CHECK, True) 1521 1522 def test_no_change_to_all_descriptors(self): 1523 base = declarative_base() 1524 1525 class Foo(base): 1526 __tablename__ = "foo" 1527 id = Column(Integer, primary_key=True) 1528 1529 eq_(Foo.__mapper__.all_orm_descriptors.keys(), ["id"]) 1530 1531 def test_oops(self): 1532 1533 with testing.expect_warnings( 1534 "Ignoring declarative-like tuple value of " "attribute 'name'" 1535 ): 1536 1537 class User(Base, fixtures.ComparableEntity): 1538 1539 __tablename__ = "users" 1540 id = Column("id", Integer, primary_key=True) 1541 name = (Column("name", String(50)),) 1542 1543 def test_table_args_no_dict(self): 1544 class Foo1(Base): 1545 1546 __tablename__ = "foo" 1547 __table_args__ = (ForeignKeyConstraint(["id"], ["foo.bar"]),) 1548 id = Column("id", Integer, primary_key=True) 1549 bar = Column("bar", Integer) 1550 1551 assert Foo1.__table__.c.id.references(Foo1.__table__.c.bar) 1552 1553 def test_table_args_type(self): 1554 def err(): 1555 class Foo1(Base): 1556 1557 __tablename__ = "foo" 1558 __table_args__ = ForeignKeyConstraint(["id"], ["foo.id"]) 1559 id = Column("id", Integer, primary_key=True) 1560 1561 assert_raises_message( 1562 sa.exc.ArgumentError, "__table_args__ value must be a tuple, ", err 1563 ) 1564 1565 def test_table_args_none(self): 1566 class Foo2(Base): 1567 1568 __tablename__ = "foo" 1569 __table_args__ = None 1570 id = Column("id", Integer, primary_key=True) 1571 1572 assert Foo2.__table__.kwargs == {} 1573 1574 def test_table_args_dict_format(self): 1575 class Foo2(Base): 1576 1577 __tablename__ = "foo" 1578 __table_args__ = {"mysql_engine": "InnoDB"} 1579 id = Column("id", Integer, primary_key=True) 1580 1581 assert Foo2.__table__.kwargs["mysql_engine"] == "InnoDB" 1582 1583 def test_table_args_tuple_format(self): 1584 class Foo2(Base): 1585 1586 __tablename__ = "foo" 1587 __table_args__ = {"mysql_engine": "InnoDB"} 1588 id = Column("id", Integer, primary_key=True) 1589 1590 class Bar(Base): 1591 1592 __tablename__ = "bar" 1593 __table_args__ = ( 1594 ForeignKeyConstraint(["id"], ["foo.id"]), 1595 {"mysql_engine": "InnoDB"}, 1596 ) 1597 id = Column("id", Integer, primary_key=True) 1598 1599 assert Bar.__table__.c.id.references(Foo2.__table__.c.id) 1600 assert Bar.__table__.kwargs["mysql_engine"] == "InnoDB" 1601 1602 def test_table_cls_attribute(self): 1603 class Foo(Base): 1604 __tablename__ = "foo" 1605 1606 @classmethod 1607 def __table_cls__(cls, *arg, **kw): 1608 name = arg[0] 1609 return Table(name + "bat", *arg[1:], **kw) 1610 1611 id = Column(Integer, primary_key=True) 1612 1613 eq_(Foo.__table__.name, "foobat") 1614 1615 def test_table_cls_attribute_return_none(self): 1616 from sqlalchemy.schema import Column, PrimaryKeyConstraint 1617 1618 class AutoTable(object): 1619 @declared_attr 1620 def __tablename__(cls): 1621 return cls.__name__ 1622 1623 @classmethod 1624 def __table_cls__(cls, *arg, **kw): 1625 for obj in arg[1:]: 1626 if ( 1627 isinstance(obj, Column) and obj.primary_key 1628 ) or isinstance(obj, PrimaryKeyConstraint): 1629 return Table(*arg, **kw) 1630 1631 return None 1632 1633 class Person(AutoTable, Base): 1634 id = Column(Integer, primary_key=True) 1635 1636 class Employee(Person): 1637 employee_name = Column(String) 1638 1639 is_(inspect(Employee).local_table, Person.__table__) 1640 1641 def test_expression(self): 1642 class User(Base, fixtures.ComparableEntity): 1643 1644 __tablename__ = "users" 1645 id = Column( 1646 "id", Integer, primary_key=True, test_needs_autoincrement=True 1647 ) 1648 name = Column("name", String(50)) 1649 addresses = relationship("Address", backref="user") 1650 1651 class Address(Base, fixtures.ComparableEntity): 1652 1653 __tablename__ = "addresses" 1654 id = Column( 1655 "id", Integer, primary_key=True, test_needs_autoincrement=True 1656 ) 1657 email = Column("email", String(50)) 1658 user_id = Column("user_id", Integer, ForeignKey("users.id")) 1659 1660 User.address_count = sa.orm.column_property( 1661 sa.select(sa.func.count(Address.id)) 1662 .where(Address.user_id == User.id) 1663 .scalar_subquery() 1664 ) 1665 Base.metadata.create_all(testing.db) 1666 u1 = User( 1667 name="u1", addresses=[Address(email="one"), Address(email="two")] 1668 ) 1669 sess = fixture_session() 1670 sess.add(u1) 1671 sess.flush() 1672 sess.expunge_all() 1673 eq_( 1674 sess.query(User).all(), 1675 [ 1676 User( 1677 name="u1", 1678 address_count=2, 1679 addresses=[Address(email="one"), Address(email="two")], 1680 ) 1681 ], 1682 ) 1683 1684 def test_useless_declared_attr(self): 1685 class Address(Base, fixtures.ComparableEntity): 1686 1687 __tablename__ = "addresses" 1688 id = Column( 1689 "id", Integer, primary_key=True, test_needs_autoincrement=True 1690 ) 1691 email = Column("email", String(50)) 1692 user_id = Column("user_id", Integer, ForeignKey("users.id")) 1693 1694 class User(Base, fixtures.ComparableEntity): 1695 1696 __tablename__ = "users" 1697 id = Column( 1698 "id", Integer, primary_key=True, test_needs_autoincrement=True 1699 ) 1700 name = Column("name", String(50)) 1701 addresses = relationship("Address", backref="user") 1702 1703 @declared_attr 1704 def address_count(cls): 1705 # this doesn't really gain us anything. but if 1706 # one is used, lets have it function as expected... 1707 return sa.orm.column_property( 1708 sa.select(sa.func.count(Address.id)) 1709 .where(Address.user_id == cls.id) 1710 .scalar_subquery() 1711 ) 1712 1713 Base.metadata.create_all(testing.db) 1714 u1 = User( 1715 name="u1", addresses=[Address(email="one"), Address(email="two")] 1716 ) 1717 sess = fixture_session() 1718 sess.add(u1) 1719 sess.flush() 1720 sess.expunge_all() 1721 eq_( 1722 sess.query(User).all(), 1723 [ 1724 User( 1725 name="u1", 1726 address_count=2, 1727 addresses=[Address(email="one"), Address(email="two")], 1728 ) 1729 ], 1730 ) 1731 1732 def test_declared_on_base_class(self): 1733 class MyBase(Base): 1734 __tablename__ = "foo" 1735 id = Column(Integer, primary_key=True) 1736 1737 @declared_attr 1738 def somecol(cls): 1739 return Column(Integer) 1740 1741 class MyClass(MyBase): 1742 __tablename__ = "bar" 1743 id = Column(Integer, ForeignKey("foo.id"), primary_key=True) 1744 1745 # previously, the 'somecol' declared_attr would be ignored 1746 # by the mapping and would remain unused. now we take 1747 # it as part of MyBase. 1748 1749 assert "somecol" in MyBase.__table__.c 1750 assert "somecol" not in MyClass.__table__.c 1751 1752 def test_decl_cascading_warns_non_mixin(self): 1753 with expect_warnings( 1754 "Use of @declared_attr.cascading only applies to " 1755 "Declarative 'mixin' and 'abstract' classes. " 1756 "Currently, this flag is ignored on mapped class " 1757 "<class '.*.MyBase'>" 1758 ): 1759 1760 class MyBase(Base): 1761 __tablename__ = "foo" 1762 id = Column(Integer, primary_key=True) 1763 1764 @declared_attr.cascading 1765 def somecol(cls): 1766 return Column(Integer) 1767 1768 def test_column(self): 1769 class User(Base, fixtures.ComparableEntity): 1770 1771 __tablename__ = "users" 1772 id = Column( 1773 "id", Integer, primary_key=True, test_needs_autoincrement=True 1774 ) 1775 name = Column("name", String(50)) 1776 1777 User.a = Column("a", String(10)) 1778 User.b = Column(String(10)) 1779 Base.metadata.create_all(testing.db) 1780 u1 = User(name="u1", a="a", b="b") 1781 eq_(u1.a, "a") 1782 eq_(User.a.get_history(u1), (["a"], (), ())) 1783 sess = fixture_session() 1784 sess.add(u1) 1785 sess.flush() 1786 sess.expunge_all() 1787 eq_(sess.query(User).all(), [User(name="u1", a="a", b="b")]) 1788 1789 def test_column_properties(self): 1790 class Address(Base, fixtures.ComparableEntity): 1791 1792 __tablename__ = "addresses" 1793 id = Column( 1794 Integer, primary_key=True, test_needs_autoincrement=True 1795 ) 1796 email = Column(String(50)) 1797 user_id = Column(Integer, ForeignKey("users.id")) 1798 1799 class User(Base, fixtures.ComparableEntity): 1800 1801 __tablename__ = "users" 1802 id = Column( 1803 "id", Integer, primary_key=True, test_needs_autoincrement=True 1804 ) 1805 name = Column("name", String(50)) 1806 1807 adr_count = sa.orm.column_property( 1808 sa.select(sa.func.count(Address.id)) 1809 .where(Address.user_id == id) 1810 .scalar_subquery() 1811 ) 1812 addresses = relationship(Address) 1813 1814 Base.metadata.create_all(testing.db) 1815 u1 = User( 1816 name="u1", addresses=[Address(email="one"), Address(email="two")] 1817 ) 1818 sess = fixture_session() 1819 sess.add(u1) 1820 sess.flush() 1821 sess.expunge_all() 1822 eq_( 1823 sess.query(User).all(), 1824 [ 1825 User( 1826 name="u1", 1827 adr_count=2, 1828 addresses=[Address(email="one"), Address(email="two")], 1829 ) 1830 ], 1831 ) 1832 1833 def test_column_properties_2(self): 1834 class Address(Base, fixtures.ComparableEntity): 1835 1836 __tablename__ = "addresses" 1837 id = Column(Integer, primary_key=True) 1838 email = Column(String(50)) 1839 user_id = Column(Integer, ForeignKey("users.id")) 1840 1841 class User(Base, fixtures.ComparableEntity): 1842 1843 __tablename__ = "users" 1844 id = Column("id", Integer, primary_key=True) 1845 name = Column("name", String(50)) 1846 1847 # this is not "valid" but we want to test that Address.id 1848 # doesn't get stuck into user's table 1849 1850 adr_count = Address.id 1851 1852 eq_(set(User.__table__.c.keys()), set(["id", "name"])) 1853 eq_(set(Address.__table__.c.keys()), set(["id", "email", "user_id"])) 1854 1855 def test_deferred(self): 1856 class User(Base, fixtures.ComparableEntity): 1857 1858 __tablename__ = "users" 1859 id = Column( 1860 Integer, primary_key=True, test_needs_autoincrement=True 1861 ) 1862 name = sa.orm.deferred(Column(String(50))) 1863 1864 Base.metadata.create_all(testing.db) 1865 sess = fixture_session() 1866 sess.add(User(name="u1")) 1867 sess.flush() 1868 sess.expunge_all() 1869 u1 = sess.query(User).filter(User.name == "u1").one() 1870 assert "name" not in u1.__dict__ 1871 1872 def go(): 1873 eq_(u1.name, "u1") 1874 1875 self.assert_sql_count(testing.db, go, 1) 1876 1877 def test_composite_inline(self): 1878 class AddressComposite(fixtures.ComparableEntity): 1879 def __init__(self, street, state): 1880 self.street = street 1881 self.state = state 1882 1883 def __composite_values__(self): 1884 return [self.street, self.state] 1885 1886 class User(Base, fixtures.ComparableEntity): 1887 __tablename__ = "user" 1888 id = Column( 1889 Integer, primary_key=True, test_needs_autoincrement=True 1890 ) 1891 address = composite( 1892 AddressComposite, 1893 Column("street", String(50)), 1894 Column("state", String(2)), 1895 ) 1896 1897 Base.metadata.create_all(testing.db) 1898 sess = fixture_session() 1899 sess.add(User(address=AddressComposite("123 anywhere street", "MD"))) 1900 sess.commit() 1901 eq_( 1902 sess.query(User).all(), 1903 [User(address=AddressComposite("123 anywhere street", "MD"))], 1904 ) 1905 1906 def test_composite_separate(self): 1907 class AddressComposite(fixtures.ComparableEntity): 1908 def __init__(self, street, state): 1909 self.street = street 1910 self.state = state 1911 1912 def __composite_values__(self): 1913 return [self.street, self.state] 1914 1915 class User(Base, fixtures.ComparableEntity): 1916 __tablename__ = "user" 1917 id = Column( 1918 Integer, primary_key=True, test_needs_autoincrement=True 1919 ) 1920 street = Column(String(50)) 1921 state = Column(String(2)) 1922 address = composite(AddressComposite, street, state) 1923 1924 Base.metadata.create_all(testing.db) 1925 sess = fixture_session() 1926 sess.add(User(address=AddressComposite("123 anywhere street", "MD"))) 1927 sess.commit() 1928 eq_( 1929 sess.query(User).all(), 1930 [User(address=AddressComposite("123 anywhere street", "MD"))], 1931 ) 1932 1933 def test_mapping_to_join(self): 1934 users = Table( 1935 "users", Base.metadata, Column("id", Integer, primary_key=True) 1936 ) 1937 addresses = Table( 1938 "addresses", 1939 Base.metadata, 1940 Column("id", Integer, primary_key=True), 1941 Column("user_id", Integer, ForeignKey("users.id")), 1942 ) 1943 usersaddresses = sa.join( 1944 users, addresses, users.c.id == addresses.c.user_id 1945 ) 1946 1947 class User(Base): 1948 __table__ = usersaddresses 1949 __table_args__ = {"primary_key": [users.c.id]} 1950 1951 # need to use column_property for now 1952 user_id = column_property(users.c.id, addresses.c.user_id) 1953 address_id = addresses.c.id 1954 1955 assert User.__mapper__.get_property("user_id").columns[0] is users.c.id 1956 assert ( 1957 User.__mapper__.get_property("user_id").columns[1] 1958 is addresses.c.user_id 1959 ) 1960 1961 def test_synonym_inline(self): 1962 class User(Base, fixtures.ComparableEntity): 1963 1964 __tablename__ = "users" 1965 id = Column( 1966 "id", Integer, primary_key=True, test_needs_autoincrement=True 1967 ) 1968 _name = Column("name", String(50)) 1969 1970 def _set_name(self, name): 1971 self._name = "SOMENAME " + name 1972 1973 def _get_name(self): 1974 return self._name 1975 1976 name = sa.orm.synonym( 1977 "_name", descriptor=property(_get_name, _set_name) 1978 ) 1979 1980 Base.metadata.create_all(testing.db) 1981 sess = fixture_session() 1982 u1 = User(name="someuser") 1983 eq_(u1.name, "SOMENAME someuser") 1984 sess.add(u1) 1985 sess.flush() 1986 eq_( 1987 sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1 1988 ) 1989 1990 def test_synonym_no_descriptor(self): 1991 from sqlalchemy.orm.properties import ColumnProperty 1992 1993 class CustomCompare(ColumnProperty.Comparator): 1994 1995 __hash__ = None 1996 1997 def __eq__(self, other): 1998 return self.__clause_element__() == other + " FOO" 1999 2000 class User(Base, fixtures.ComparableEntity): 2001 2002 __tablename__ = "users" 2003 id = Column( 2004 "id", Integer, primary_key=True, test_needs_autoincrement=True 2005 ) 2006 _name = Column("name", String(50)) 2007 name = sa.orm.synonym("_name", comparator_factory=CustomCompare) 2008 2009 Base.metadata.create_all(testing.db) 2010 sess = fixture_session() 2011 u1 = User(name="someuser FOO") 2012 sess.add(u1) 2013 sess.flush() 2014 eq_(sess.query(User).filter(User.name == "someuser").one(), u1) 2015 2016 def test_synonym_added(self): 2017 class User(Base, fixtures.ComparableEntity): 2018 2019 __tablename__ = "users" 2020 id = Column( 2021 "id", Integer, primary_key=True, test_needs_autoincrement=True 2022 ) 2023 _name = Column("name", String(50)) 2024 2025 def _set_name(self, name): 2026 self._name = "SOMENAME " + name 2027 2028 def _get_name(self): 2029 return self._name 2030 2031 name = property(_get_name, _set_name) 2032 2033 User.name = sa.orm.synonym("_name", descriptor=User.name) 2034 Base.metadata.create_all(testing.db) 2035 sess = fixture_session() 2036 u1 = User(name="someuser") 2037 eq_(u1.name, "SOMENAME someuser") 2038 sess.add(u1) 2039 sess.flush() 2040 eq_( 2041 sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1 2042 ) 2043 2044 def test_reentrant_compile_via_foreignkey(self): 2045 class User(Base, fixtures.ComparableEntity): 2046 2047 __tablename__ = "users" 2048 id = Column( 2049 "id", Integer, primary_key=True, test_needs_autoincrement=True 2050 ) 2051 name = Column("name", String(50)) 2052 addresses = relationship("Address", backref="user") 2053 2054 class Address(Base, fixtures.ComparableEntity): 2055 2056 __tablename__ = "addresses" 2057 id = Column( 2058 "id", Integer, primary_key=True, test_needs_autoincrement=True 2059 ) 2060 email = Column("email", String(50)) 2061 user_id = Column("user_id", Integer, ForeignKey(User.id)) 2062 2063 # previous versions would force a re-entrant mapper compile via 2064 # the User.id inside the ForeignKey but this is no longer the 2065 # case 2066 2067 sa.orm.configure_mappers() 2068 eq_( 2069 list(Address.user_id.property.columns[0].foreign_keys)[0].column, 2070 User.__table__.c.id, 2071 ) 2072 Base.metadata.create_all(testing.db) 2073 u1 = User( 2074 name="u1", addresses=[Address(email="one"), Address(email="two")] 2075 ) 2076 sess = fixture_session() 2077 sess.add(u1) 2078 sess.flush() 2079 sess.expunge_all() 2080 eq_( 2081 sess.query(User).all(), 2082 [ 2083 User( 2084 name="u1", 2085 addresses=[Address(email="one"), Address(email="two")], 2086 ) 2087 ], 2088 ) 2089 2090 def test_relationship_reference(self): 2091 class Address(Base, fixtures.ComparableEntity): 2092 2093 __tablename__ = "addresses" 2094 id = Column( 2095 "id", Integer, primary_key=True, test_needs_autoincrement=True 2096 ) 2097 email = Column("email", String(50)) 2098 user_id = Column("user_id", Integer, ForeignKey("users.id")) 2099 2100 class User(Base, fixtures.ComparableEntity): 2101 2102 __tablename__ = "users" 2103 id = Column( 2104 "id", Integer, primary_key=True, test_needs_autoincrement=True 2105 ) 2106 name = Column("name", String(50)) 2107 addresses = relationship( 2108 "Address", backref="user", primaryjoin=id == Address.user_id 2109 ) 2110 2111 User.address_count = sa.orm.column_property( 2112 sa.select(sa.func.count(Address.id)) 2113 .where(Address.user_id == User.id) 2114 .scalar_subquery() 2115 ) 2116 Base.metadata.create_all(testing.db) 2117 u1 = User( 2118 name="u1", addresses=[Address(email="one"), Address(email="two")] 2119 ) 2120 sess = fixture_session() 2121 sess.add(u1) 2122 sess.flush() 2123 sess.expunge_all() 2124 eq_( 2125 sess.query(User).all(), 2126 [ 2127 User( 2128 name="u1", 2129 address_count=2, 2130 addresses=[Address(email="one"), Address(email="two")], 2131 ) 2132 ], 2133 ) 2134 2135 def test_pk_with_fk_init(self): 2136 class Bar(Base): 2137 2138 __tablename__ = "bar" 2139 id = sa.Column( 2140 sa.Integer, sa.ForeignKey("foo.id"), primary_key=True 2141 ) 2142 ex = sa.Column(sa.Integer, primary_key=True) 2143 2144 class Foo(Base): 2145 2146 __tablename__ = "foo" 2147 id = sa.Column(sa.Integer, primary_key=True) 2148 bars = sa.orm.relationship(Bar) 2149 2150 assert Bar.__mapper__.primary_key[0] is Bar.__table__.c.id 2151 assert Bar.__mapper__.primary_key[1] is Bar.__table__.c.ex 2152 2153 @testing.provide_metadata 2154 def test_with_explicit_autoloaded(self): 2155 meta = self.metadata 2156 t1 = Table( 2157 "t1", 2158 meta, 2159 Column("id", String(50), primary_key=True), 2160 Column("data", String(50)), 2161 ) 2162 meta.create_all(testing.db) 2163 2164 class MyObj(Base): 2165 2166 __table__ = Table("t1", Base.metadata, autoload_with=testing.db) 2167 2168 sess = fixture_session() 2169 m = MyObj(id="someid", data="somedata") 2170 sess.add(m) 2171 sess.flush() 2172 eq_(sess.execute(t1.select()).fetchall(), [("someid", "somedata")]) 2173 2174 def test_synonym_for(self): 2175 class User(Base, fixtures.ComparableEntity): 2176 2177 __tablename__ = "users" 2178 id = Column( 2179 "id", Integer, primary_key=True, test_needs_autoincrement=True 2180 ) 2181 name = Column("name", String(50)) 2182 2183 @synonym_for("name") 2184 @property 2185 def namesyn(self): 2186 return self.name 2187 2188 Base.metadata.create_all(testing.db) 2189 sess = fixture_session() 2190 u1 = User(name="someuser") 2191 eq_(u1.name, "someuser") 2192 eq_(u1.namesyn, "someuser") 2193 sess.add(u1) 2194 sess.flush() 2195 rt = sess.query(User).filter(User.namesyn == "someuser").one() 2196 eq_(rt, u1) 2197 2198 def test_duplicate_classes_in_base(self): 2199 class Test(Base): 2200 __tablename__ = "a" 2201 id = Column(Integer, primary_key=True) 2202 2203 assert_raises_message( 2204 sa.exc.SAWarning, 2205 "This declarative base already contains a class with ", 2206 lambda: type(Base)( 2207 "Test", 2208 (Base,), 2209 dict(__tablename__="b", id=Column(Integer, primary_key=True)), 2210 ), 2211 ) 2212 2213 @testing.teardown_events(MapperEvents) 2214 @testing.teardown_events(InstrumentationEvents) 2215 def test_instrument_class_before_instrumentation(self): 2216 # test #3388 2217 2218 canary = mock.Mock() 2219 2220 @event.listens_for(mapper, "instrument_class") 2221 def instrument_class(mp, cls): 2222 canary.instrument_class(mp, cls) 2223 2224 @event.listens_for(object, "class_instrument") 2225 def class_instrument(cls): 2226 canary.class_instrument(cls) 2227 2228 class Test(Base): 2229 __tablename__ = "test" 2230 id = Column(Integer, primary_key=True) 2231 2232 eq_( 2233 canary.mock_calls, 2234 [ 2235 mock.call.instrument_class(Test.__mapper__, Test), 2236 mock.call.class_instrument(Test), 2237 ], 2238 ) 2239 2240 def test_cls_docstring(self): 2241 class MyBase(object): 2242 """MyBase Docstring""" 2243 2244 Base = declarative_base(cls=MyBase) 2245 2246 eq_(Base.__doc__, MyBase.__doc__) 2247 2248 def test_delattr_mapped_raises(self): 2249 Base = declarative_base() 2250 2251 class Foo(Base): 2252 __tablename__ = "foo" 2253 2254 id = Column(Integer, primary_key=True) 2255 data = Column(String) 2256 2257 def go(): 2258 del Foo.data 2259 2260 assert_raises_message( 2261 NotImplementedError, 2262 "Can't un-map individual mapped attributes on a mapped class.", 2263 go, 2264 ) 2265 2266 def test_delattr_hybrid_fine(self): 2267 Base = declarative_base() 2268 2269 class Foo(Base): 2270 __tablename__ = "foo" 2271 2272 id = Column(Integer, primary_key=True) 2273 data = Column(String) 2274 2275 @hybrid_property 2276 def data_hybrid(self): 2277 return self.data 2278 2279 assert "data_hybrid" in Foo.__mapper__.all_orm_descriptors.keys() 2280 2281 del Foo.data_hybrid 2282 2283 assert "data_hybrid" not in Foo.__mapper__.all_orm_descriptors.keys() 2284 2285 assert not hasattr(Foo, "data_hybrid") 2286 2287 def test_setattr_hybrid_updates_descriptors(self): 2288 Base = declarative_base() 2289 2290 class Foo(Base): 2291 __tablename__ = "foo" 2292 2293 id = Column(Integer, primary_key=True) 2294 data = Column(String) 2295 2296 assert "data_hybrid" not in Foo.__mapper__.all_orm_descriptors.keys() 2297 2298 @hybrid_property 2299 def data_hybrid(self): 2300 return self.data 2301 2302 Foo.data_hybrid = data_hybrid 2303 assert "data_hybrid" in Foo.__mapper__.all_orm_descriptors.keys() 2304 2305 del Foo.data_hybrid 2306 2307 assert "data_hybrid" not in Foo.__mapper__.all_orm_descriptors.keys() 2308 2309 assert not hasattr(Foo, "data_hybrid") 2310 2311 def test_classes_can_override_new(self): 2312 class MyTable(Base): 2313 __tablename__ = "my_table" 2314 id = Column(Integer, primary_key=True) 2315 2316 def __new__(cls, *args, **kwargs): 2317 return object.__new__(cls) 2318 2319 def some_method(self): 2320 pass 2321 2322 @staticmethod 2323 def some_static_method(self): 2324 pass 2325 2326 mt = MyTable(id=5) 2327 eq_(mt.id, 5) 2328 2329 @testing.requires.python36 2330 def test_kw_support_in_declarative_meta_init(self): 2331 # This will not fail if DeclarativeMeta __init__ supports **kw 2332 2333 class BaseUser(Base): 2334 __tablename__ = "base" 2335 id_ = Column(Integer, primary_key=True) 2336 2337 @classmethod 2338 def __init_subclass__(cls, random_keyword=False, **kw): 2339 super().__init_subclass__(**kw) 2340 cls._set_random_keyword_used_here = random_keyword 2341 2342 class User(BaseUser): 2343 __tablename__ = "user" 2344 id_ = Column(Integer, ForeignKey("base.id_"), primary_key=True) 2345 2346 # Check the default option 2347 eq_(User._set_random_keyword_used_here, False) 2348 2349 # Build the metaclass with a keyword! 2350 bases = (BaseUser,) 2351 UserType = DeclarativeMeta("UserType", bases, {}, random_keyword=True) 2352 2353 # Check to see if __init_subclass__ works in supported versions 2354 eq_(UserType._set_random_keyword_used_here, True) 2355